Skip to content

cdm_data_loaders.core.registry

Validated registries and module loaders for entity models, JSON Schemas, PyArrow schemas, transforms, and pre-processors.

RejectedFreezerMixin pydantic-model

Bases: BaseModel

Mixin to forbid the use of the reserved _rejected table name in registries.

Show JSON schema: 
{
  "description": "Mixin to forbid the use of the reserved ``_rejected`` table name in registries.",
  "properties": {},
  "title": "RejectedFreezerMixin",
  "type": "object"
}

Validators:

  • _validate_and_freeze
Source code in src/cdm_data_loaders/core/registry.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class RejectedFreezerMixin(BaseModel):
    """Mixin to forbid the use of the reserved ``_rejected`` table name in registries."""

    @model_validator(mode="after")
    def _validate_and_freeze(self) -> Self:
        """Forbid ``_rejected`` in registry keys."""
        if REJECTED_TABLE_NAME in self.registry:  # pyright: ignore[reportAttributeAccessIssue]
            err_msg = (
                f"{self.__class__.__name__} cannot use reserved table name {REJECTED_TABLE_NAME!r}; "
                f"{REJECTED_TABLE_NAME!r} is reserved for pipeline error records."
            )
            raise ValueError(err_msg)
        object.__setattr__(self, "registry", MappingProxyType(dict(self.registry)))  # pyright: ignore[reportAttributeAccessIssue]
        return self

ModelRegistry pydantic-model

Bases: RejectedFreezerMixin, BaseModel

A validated registry of entity models.

Show JSON schema: 
{
  "description": "A validated registry of entity models.",
  "properties": {
    "registry": {
      "additionalProperties": true,
      "minProperties": 1,
      "propertyNames": {
        "minLength": 1
      },
      "title": "Registry",
      "type": "object"
    }
  },
  "required": [
    "registry"
  ],
  "title": "ModelRegistry",
  "type": "object"
}

Config:

  • str_strip_whitespace: True

Fields:

  • registry (dict[RegistryKey, type[BaseModel]])

Validators:

  • _validate_and_freeze
Source code in src/cdm_data_loaders/core/registry.py 
61
62
63
64
65
66
class ModelRegistry(RejectedFreezerMixin, BaseModel):
    """A validated registry of entity models."""

    model_config = ConfigDict(str_strip_whitespace=True)

    registry: Annotated[dict[RegistryKey, type[BaseModel]], Field(min_length=1)]

SchemaRegistry pydantic-model

Bases: RejectedFreezerMixin, BaseModel

A validated registry of validated entity schemas.

Show JSON schema: 
{
  "$defs": {
    "ValidatedSchema": {
      "description": "A validated JSON Schema.",
      "properties": {
        "jsonschema": {
          "additionalProperties": true,
          "title": "Jsonschema",
          "type": "object"
        }
      },
      "required": [
        "jsonschema"
      ],
      "title": "ValidatedSchema",
      "type": "object"
    }
  },
  "description": "A validated registry of validated entity schemas.",
  "properties": {
    "registry": {
      "additionalProperties": {
        "$ref": "#/$defs/ValidatedSchema"
      },
      "minProperties": 1,
      "propertyNames": {
        "minLength": 1
      },
      "title": "Registry",
      "type": "object"
    }
  },
  "required": [
    "registry"
  ],
  "title": "SchemaRegistry",
  "type": "object"
}

Config:

  • str_strip_whitespace: True

Fields:

  • registry (dict[RegistryKey, Annotated[ValidatedSchema, BeforeValidator(_coerce_validated_schema)]])

Validators:

  • _validate_and_freeze
Source code in src/cdm_data_loaders/core/registry.py 
69
70
71
72
73
74
75
76
77
class SchemaRegistry(RejectedFreezerMixin, BaseModel):
    """A validated registry of validated entity schemas."""

    model_config = ConfigDict(str_strip_whitespace=True)

    registry: Annotated[
        dict[RegistryKey, Annotated[ValidatedSchema, BeforeValidator(_coerce_validated_schema)]],
        Field(min_length=1),
    ]

ArrowSchemaRegistry pydantic-model

Bases: RejectedFreezerMixin, BaseModel

A validated registry of PyArrow schemas.

Config:

  • frozen: True
  • str_strip_whitespace: True

Fields:

  • registry (dict[RegistryKey, InstanceOf[Schema]])

Validators:

  • _validate_and_freeze
Source code in src/cdm_data_loaders/core/registry.py 
80
81
82
83
84
85
86
87
88
class ArrowSchemaRegistry(RejectedFreezerMixin, BaseModel):
    """A validated registry of PyArrow schemas."""

    model_config = ConfigDict(frozen=True, str_strip_whitespace=True)

    registry: Annotated[
        dict[RegistryKey, InstanceOf[pa.Schema]],
        Field(min_length=1),
    ]

TransformRegistry pydantic-model

Bases: RejectedFreezerMixin, BaseModel

A validated registry of per-table record transform functions.

Config:

  • frozen: True
  • str_strip_whitespace: True

Fields:

  • registry (dict[RegistryKey, RecordTransformFn])

Validators:

  • _validate_and_freeze
Source code in src/cdm_data_loaders/core/registry.py 
91
92
93
94
95
96
class TransformRegistry(RejectedFreezerMixin, BaseModel):
    """A validated registry of per-table record transform functions."""

    model_config = ConfigDict(frozen=True, str_strip_whitespace=True)

    registry: Annotated[dict[RegistryKey, RecordTransformFn], Field(min_length=1)]

PreprocessorRegistry pydantic-model

Bases: RejectedFreezerMixin, BaseModel

A validated registry of pipeline-wide or per-table file pre-processor functions.

Config:

  • frozen: True
  • str_strip_whitespace: True

Fields:

  • default (tuple[FilePreprocessFn, ...])
  • registry (dict[RegistryKey, PreprocessorTuple])

Validators:

  • _validate_and_freeze
  • _validate_mode
Source code in src/cdm_data_loaders/core/registry.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class PreprocessorRegistry(RejectedFreezerMixin, BaseModel):
    """A validated registry of pipeline-wide or per-table file pre-processor functions."""

    model_config = ConfigDict(frozen=True, str_strip_whitespace=True)

    default: tuple[FilePreprocessFn, ...] = ()
    registry: dict[RegistryKey, PreprocessorTuple] = Field(default_factory=dict)

    @model_validator(mode="after")
    def _validate_mode(self) -> Self:
        """Require either ``default`` or ``registry`` to be populated, not both."""
        if not self.default and not self.registry:
            err_msg = "PreprocessorRegistry must define at least one preprocessor in 'default' or 'registry'."
            raise ValueError(err_msg)
        if self.default and self.registry:
            err_msg = "PreprocessorRegistry cannot define both 'default' and 'registry' preprocessors."
            raise ValueError(err_msg)
        return self

    def get(self, table_name: str | None = None) -> tuple[FilePreprocessFn, ...]:
        """Return the file pre-processors configured for ``table_name`` or pipeline-wide default.

        :param table_name: target table name, or ``None`` for multi-table streams
        :type table_name: str | None
        :return: tuple of file pre-processor callables to execute in order
        :rtype: tuple[FilePreprocessFn, ...]
        """
        if self.default:
            return self.default
        if table_name is not None:
            return self.registry.get(table_name, ())
        seen: list[FilePreprocessFn] = []
        for funcs in self.registry.values():
            for fn in funcs:
                if fn not in seen:
                    seen.append(fn)
        return tuple(seen)

get 

get(
    table_name: str | None = None,
) -> tuple[FilePreprocessFn, ...]

Return the file pre-processors configured for table_name or pipeline-wide default.

Parameters:

Name Type Description Default
table_name str | None

target table name, or None for multi-table streams

 None

Returns:

Type Description
tuple[FilePreprocessFn, ...]

tuple of file pre-processor callables to execute in order
Source code in src/cdm_data_loaders/core/registry.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def get(self, table_name: str | None = None) -> tuple[FilePreprocessFn, ...]:
    """Return the file pre-processors configured for ``table_name`` or pipeline-wide default.

    :param table_name: target table name, or ``None`` for multi-table streams
    :type table_name: str | None
    :return: tuple of file pre-processor callables to execute in order
    :rtype: tuple[FilePreprocessFn, ...]
    """
    if self.default:
        return self.default
    if table_name is not None:
        return self.registry.get(table_name, ())
    seen: list[FilePreprocessFn] = []
    for funcs in self.registry.values():
        for fn in funcs:
            if fn not in seen:
                seen.append(fn)
    return tuple(seen)

load_registry_dict 

load_registry_dict(
    module_path: str,
    attribute_name: Literal["ENTITY_MODELS"],
) -> ModelRegistry

load_registry_dict(
    module_path: str,
    attribute_name: Literal["ENTITY_SCHEMAS"],
) -> SchemaRegistry

load_registry_dict(
    module_path: str,
    attribute_name: Literal["TABLE_TRANSFORMS"],
) -> TransformRegistry

load_registry_dict(
    module_path: str,
    attribute_name: Literal["OUTPUT_SCHEMAS"],
) -> ArrowSchemaRegistry

load_registry_dict(
    module_path: str, attribute_name: str
) -> (
    ModelRegistry
    | SchemaRegistry
    | TransformRegistry
    | ArrowSchemaRegistry
)

Import module_path and return its validated registry for attribute_name.

Parameters:

Name Type Description Default
module_path str

dotted import path of the registry module

 required
attribute_name str

name of the dictionary attribute on the module (ENTITY_MODELS, ENTITY_SCHEMAS, TABLE_TRANSFORMS, or OUTPUT_SCHEMAS)

 required

Returns:

Type Description
ModelRegistry | SchemaRegistry | TransformRegistry | ArrowSchemaRegistry

the validated registry instance

Raises:

Type Description
ValueError

if the attribute name is unsupported or uses the reserved _rejected table name
Source code in src/cdm_data_loaders/core/registry.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def load_registry_dict(
    module_path: str,
    attribute_name: str,
) -> ModelRegistry | SchemaRegistry | TransformRegistry | ArrowSchemaRegistry:
    """Import ``module_path`` and return its validated registry for ``attribute_name``.

    :param module_path: dotted import path of the registry module
    :type module_path: str
    :param attribute_name: name of the dictionary attribute on the module
        (``ENTITY_MODELS``, ``ENTITY_SCHEMAS``, ``TABLE_TRANSFORMS``, or ``OUTPUT_SCHEMAS``)
    :type attribute_name: str
    :return: the validated registry instance
    :rtype: ModelRegistry | SchemaRegistry | TransformRegistry | ArrowSchemaRegistry
    :raises ValueError: if the attribute name is unsupported or uses the reserved ``_rejected`` table name
    """
    module = import_module(module_path)
    raw_registry = getattr(module, attribute_name, None)
    if not isinstance(raw_registry, dict):
        err_msg = f"{module_path}.{attribute_name} must be a dictionary, not {type(raw_registry).__name__}"
        raise TypeError(err_msg)

    cls_map = {
        "ENTITY_MODELS": ModelRegistry,
        "ENTITY_SCHEMAS": SchemaRegistry,
        "TABLE_TRANSFORMS": TransformRegistry,
        "OUTPUT_SCHEMAS": ArrowSchemaRegistry,
    }
    if attribute_name not in cls_map:
        err_msg = f"Unsupported registry attribute: {attribute_name}"
        raise ValueError(err_msg)

    return cls_map[attribute_name](registry=raw_registry)  # pyright: ignore[reportArgumentType]

load_file_preprocessors 

load_file_preprocessors(
    module_path: str,
    attribute_name: str = DEFAULT_PREPROCESSORS_ATTR,
) -> PreprocessorRegistry

Import module_path and return its validated :class:PreprocessorRegistry.

Parameters:

Name Type Description Default
module_path str

dotted import path of the pre-processors module

 required
attribute_name str

name of the attribute on the module (defaults to FILE_PREPROCESSORS)

 DEFAULT_PREPROCESSORS_ATTR

Returns:

Type Description
PreprocessorRegistry

the validated file pre-processor registry

Raises:

Type Description
TypeError

if the attribute is neither a dictionary nor a sequence of callables
ValueError

if the registry is empty or uses the reserved _rejected table name
Source code in src/cdm_data_loaders/core/registry.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def load_file_preprocessors(
    module_path: str,
    attribute_name: str = DEFAULT_PREPROCESSORS_ATTR,
) -> PreprocessorRegistry:
    """Import ``module_path`` and return its validated :class:`PreprocessorRegistry`.

    :param module_path: dotted import path of the pre-processors module
    :type module_path: str
    :param attribute_name: name of the attribute on the module (defaults to ``FILE_PREPROCESSORS``)
    :type attribute_name: str
    :return: the validated file pre-processor registry
    :rtype: PreprocessorRegistry
    :raises TypeError: if the attribute is neither a dictionary nor a sequence of callables
    :raises ValueError: if the registry is empty or uses the reserved ``_rejected`` table name
    """
    module = import_module(module_path)
    raw_preprocessors = getattr(module, attribute_name, None)
    if isinstance(raw_preprocessors, dict):
        return PreprocessorRegistry(registry=raw_preprocessors)
    if isinstance(raw_preprocessors, Sequence) and not isinstance(raw_preprocessors, (str, bytes)):
        return PreprocessorRegistry(default=tuple(raw_preprocessors))
    err_msg = (
        f"{module_path}.{attribute_name} must be a dictionary or a sequence of callables, "
        f"not {type(raw_preprocessors).__name__}"
    )
    raise TypeError(err_msg)

validate_registry_table_names 

validate_registry_table_names(
    registry_keys: Iterable[str],
    valid_tables: Iterable[str],
    module_label: str,
) -> None

Raise ValueError if registry_keys contains table names not in valid_tables.

Parameters:

Name Type Description Default
registry_keys Iterable[str]

table names configured in a registry

 required
valid_tables Iterable[str]

valid destination/input table names for the pipeline

 required
module_label str

setting attribute label reported in the error message

 required

Raises:

Type Description
ValueError

if registry_keys contains unknown table names
Source code in src/cdm_data_loaders/core/registry.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def validate_registry_table_names(
    registry_keys: Iterable[str],
    valid_tables: Iterable[str],
    module_label: str,
) -> None:
    """Raise ``ValueError`` if ``registry_keys`` contains table names not in ``valid_tables``.

    :param registry_keys: table names configured in a registry
    :type registry_keys: Iterable[str]
    :param valid_tables: valid destination/input table names for the pipeline
    :type valid_tables: Iterable[str]
    :param module_label: setting attribute label reported in the error message
    :type module_label: str
    :raises ValueError: if ``registry_keys`` contains unknown table names
    """
    unknown_tables = sorted(set(registry_keys) - set(valid_tables))
    if unknown_tables:
        err_msg = f"Unknown table names in {module_label}: {', '.join(unknown_tables)}"
        raise ValueError(err_msg)