Skip to content

dqm_ml_core.models.global_

Global configuration models for storage, compute, and error handling.

Defines shared configuration classes used across pipeline interfaces: - RetryConfig: Retry policies for storage operations - StorageConfig: S3/local storage with credential management - ComputeConfig: Runtime settings (device, memory, threads, logging) - ImageErrorsConfig/TabularErrorsConfig/ErrorsConfig: Error handling policies

ComputeConfig

Bases: BaseModel

Global compute / runtime settings.

Source code in packages/dqm-ml-core/src/dqm_ml_core/models/global_.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class ComputeConfig(BaseModel):
    """Global compute / runtime settings."""

    model_config = ConfigDict(extra="forbid")

    seed: int = Field(default=42, description="Random seed for reproducibility.")
    log_level: Literal["debug", "info", "warning", "error"] = Field(
        default="warning",
        description="Logging verbosity.",
    )
    max_memory: str | None = Field(default=None, description="Maximum memory per worker (e.g. '4Gi').")
    device: Literal["auto", "cpu", "cuda"] = Field(
        default="auto",
        description="Compute device: 'auto' picks cuda if available.",
    )
    progress_bar: bool = Field(default=True, description="Show tqdm progress bars.")
    threads: int = Field(default=4, gt=0, description="Number of worker threads.")

device: Literal['auto', 'cpu', 'cuda'] = Field(default='auto', description="Compute device: 'auto' picks cuda if available.") class-attribute instance-attribute

log_level: Literal['debug', 'info', 'warning', 'error'] = Field(default='warning', description='Logging verbosity.') class-attribute instance-attribute

max_memory: str | None = Field(default=None, description="Maximum memory per worker (e.g. '4Gi').") class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

progress_bar: bool = Field(default=True, description='Show tqdm progress bars.') class-attribute instance-attribute

seed: int = Field(default=42, description='Random seed for reproducibility.') class-attribute instance-attribute

threads: int = Field(default=4, gt=0, description='Number of worker threads.') class-attribute instance-attribute

ErrorsConfig

Bases: BaseModel

Aggregate error-handling configuration.

Source code in packages/dqm-ml-core/src/dqm_ml_core/models/global_.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class ErrorsConfig(BaseModel):
    """Aggregate error-handling configuration."""

    model_config = ConfigDict(extra="forbid")

    default: Literal["silent_fail", "fail_fast"] = Field(
        default="silent_fail",
        description="Default error action when no specific policy is set.",
    )
    images: ImageErrorsConfig | None = None
    tabular: TabularErrorsConfig | None = None
    max_failure_rate: float = Field(
        default=0.05,
        ge=0,
        le=1,
        description="Maximum tolerated failure rate before the job aborts (from 0.0 to 1.0).",
    )

default: Literal['silent_fail', 'fail_fast'] = Field(default='silent_fail', description='Default error action when no specific policy is set.') class-attribute instance-attribute

images: ImageErrorsConfig | None = None class-attribute instance-attribute

max_failure_rate: float = Field(default=0.05, ge=0, le=1, description='Maximum tolerated failure rate before the job aborts (from 0.0 to 1.0).') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

tabular: TabularErrorsConfig | None = None class-attribute instance-attribute

ImageErrorsConfig

Bases: BaseModel

Error-handling policy for image-processing failures.

Source code in packages/dqm-ml-core/src/dqm_ml_core/models/global_.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class ImageErrorsConfig(BaseModel):
    """Error-handling policy for image-processing failures."""

    model_config = ConfigDict(extra="forbid")

    on_decode_failure: Literal["silent_fail", "fail_fast"] = Field(
        default="silent_fail",
        description="Action when an image cannot be decoded.",
    )
    on_transform_error: Literal["silent_fail", "fail_fast"] = Field(
        default="silent_fail",
        description="Action when an image transform fails.",
    )
    on_unsupported_format: Literal["silent_fail", "fail_fast"] = Field(
        default="fail_fast",
        description="Action on unsupported image format.",
    )

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

on_decode_failure: Literal['silent_fail', 'fail_fast'] = Field(default='silent_fail', description='Action when an image cannot be decoded.') class-attribute instance-attribute

on_transform_error: Literal['silent_fail', 'fail_fast'] = Field(default='silent_fail', description='Action when an image transform fails.') class-attribute instance-attribute

on_unsupported_format: Literal['silent_fail', 'fail_fast'] = Field(default='fail_fast', description='Action on unsupported image format.') class-attribute instance-attribute

RetryConfig

Bases: BaseModel

Retry policy for storage operations.

Source code in packages/dqm-ml-core/src/dqm_ml_core/models/global_.py 
15
16
17
18
19
20
21
22
23
24
class RetryConfig(BaseModel):
    """Retry policy for storage operations."""

    model_config = ConfigDict(extra="forbid")

    mode: Literal["default", "standard"] = Field(
        default="standard",
        description="Retry mode: 'default' uses exponential backoff, 'standard' uses fixed intervals.",
    )
    max_attempts: int = Field(default=3, gt=0, description="Maximum number of retry attempts.")

max_attempts: int = Field(default=3, gt=0, description='Maximum number of retry attempts.') class-attribute instance-attribute

mode: Literal['default', 'standard'] = Field(default='standard', description="Retry mode: 'default' uses exponential backoff, 'standard' uses fixed intervals.") class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

StorageConfig

Bases: BaseModel

Remote or local storage configuration.

Source code in packages/dqm-ml-core/src/dqm_ml_core/models/global_.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class StorageConfig(BaseModel):
    """Remote or local storage configuration."""

    model_config = ConfigDict(extra="forbid")

    type: Literal["s3", "local"] = Field(description="Storage backend type.")
    bucket: str | None = Field(default=None, description="S3 bucket name (required when type='s3').")

    access_key: str | None = Field(default=None, description="AWS access key ID.")
    secret_key: str | None = Field(default=None, description="AWS secret access key.")
    session_token: str | None = Field(default=None, description="AWS session token.")
    anonymous: bool = Field(default=False, description="Use anonymous (unsigned) requests.")
    role_arn: str | None = Field(default=None, description="ARN of IAM role to assume.")
    session_name: str | None = Field(default=None, description="Name for the assumed role session.")
    external_id: str | None = Field(default=None, description="External ID for role assumption.")
    load_frequency: int = Field(default=900, gt=0, description="Frequency (seconds) to refresh credentials / role.")

    region: str | None = Field(default=None, description="AWS region (e.g. 'us-east-1').")
    endpoint: str | None = Field(default=None, description="Custom S3 endpoint URL.")
    request_timeout: float | None = Field(default=None, description="Request timeout in seconds.")
    connect_timeout: float | None = Field(default=None, description="Connection timeout in seconds.")
    scheme: str | None = Field(default=None, description="URI scheme (e.g. 'https').")
    proxy_options: dict[str, Any] | str | None = Field(
        default=None,
        description="Proxy configuration as a dict or URL string.",
    )
    tls_ca_file_path: str | None = Field(default=None, description="Path to a custom TLS CA bundle.")

    retry: RetryConfig | None = None

    checksum_validation: Literal["when_required", "always", "never"] = Field(
        default="when_required",
        description="Controls S3 checksum validation behaviour.",
    )

    @model_validator(mode="after")
    def _require_bucket_for_s3(self) -> "StorageConfig":
        """Validate that S3 storage configuration includes a bucket name.

        Returns:
            The validated StorageConfig instance.

        Raises:
            ValueError: If storage type is 's3' but no bucket is provided.
        """
        if self.type == "s3" and self.bucket is None:
            raise ValueError("StorageConfig with type='s3' requires a 'bucket'")
        return self

access_key: str | None = Field(default=None, description='AWS access key ID.') class-attribute instance-attribute

anonymous: bool = Field(default=False, description='Use anonymous (unsigned) requests.') class-attribute instance-attribute

bucket: str | None = Field(default=None, description="S3 bucket name (required when type='s3').") class-attribute instance-attribute

checksum_validation: Literal['when_required', 'always', 'never'] = Field(default='when_required', description='Controls S3 checksum validation behaviour.') class-attribute instance-attribute

connect_timeout: float | None = Field(default=None, description='Connection timeout in seconds.') class-attribute instance-attribute

endpoint: str | None = Field(default=None, description='Custom S3 endpoint URL.') class-attribute instance-attribute

external_id: str | None = Field(default=None, description='External ID for role assumption.') class-attribute instance-attribute

load_frequency: int = Field(default=900, gt=0, description='Frequency (seconds) to refresh credentials / role.') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

proxy_options: dict[str, Any] | str | None = Field(default=None, description='Proxy configuration as a dict or URL string.') class-attribute instance-attribute

region: str | None = Field(default=None, description="AWS region (e.g. 'us-east-1').") class-attribute instance-attribute

request_timeout: float | None = Field(default=None, description='Request timeout in seconds.') class-attribute instance-attribute

retry: RetryConfig | None = None class-attribute instance-attribute

role_arn: str | None = Field(default=None, description='ARN of IAM role to assume.') class-attribute instance-attribute

scheme: str | None = Field(default=None, description="URI scheme (e.g. 'https').") class-attribute instance-attribute

secret_key: str | None = Field(default=None, description='AWS secret access key.') class-attribute instance-attribute

session_name: str | None = Field(default=None, description='Name for the assumed role session.') class-attribute instance-attribute

session_token: str | None = Field(default=None, description='AWS session token.') class-attribute instance-attribute

tls_ca_file_path: str | None = Field(default=None, description='Path to a custom TLS CA bundle.') class-attribute instance-attribute

type: Literal['s3', 'local'] = Field(description='Storage backend type.') class-attribute instance-attribute

TabularErrorsConfig

Bases: BaseModel

Error-handling policy for tabular-data failures.

Source code in packages/dqm-ml-core/src/dqm_ml_core/models/global_.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
class TabularErrorsConfig(BaseModel):
    """Error-handling policy for tabular-data failures."""

    model_config = ConfigDict(extra="forbid")

    on_missing_column: Literal["silent_fail", "fail_fast"] = Field(
        default="fail_fast",
        description="Action when a required column is missing.",
    )
    on_file_not_found: Literal["silent_fail", "fail_fast"] = Field(
        default="fail_fast",
        description="Action when a data file is not found.",
    )

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

on_file_not_found: Literal['silent_fail', 'fail_fast'] = Field(default='fail_fast', description='Action when a data file is not found.') class-attribute instance-attribute

on_missing_column: Literal['silent_fail', 'fail_fast'] = Field(default='fail_fast', description='Action when a required column is missing.') class-attribute instance-attribute