cdm_data_loaders.pipelines.jsonlines.settings ¶

class JsonlJsonschemaError(RuntimeError):
    """Base error raised when an ``ENTITY_SCHEMAS`` registry or entity JSON Schema is invalid or cannot convert to PyArrow."""

    code: ClassVar[str] = ""
    title: ClassVar[str] = ""

class InvalidEntitySchemasRegistryError(JsonlJsonschemaError):
    """The `entity_schemas` module must define a non-empty `ENTITY_SCHEMAS` dictionary and cannot use the reserved `_rejected` table name."""

    code: ClassVar[str] = "E1"
    title: ClassVar[str] = "Invalid or empty ENTITY_SCHEMAS registry"

class UnsupportedJsonlSchemaDialectError(JsonlJsonschemaError):
    """Every entity JSON Schema must declare a supported `$schema` draft URI (Draft 4, 6, 7, 2019-09, 2020-12, or KBase xsv_config)."""

    code: ClassVar[str] = "E2"
    title: ClassVar[str] = "Missing or unsupported JSON Schema $schema dialect"

class InvalidJsonlEntitySchemaError(JsonlJsonschemaError):
    """Every entity JSON Schema must be a valid schema document under its declared `$schema` metaschema."""

    code: ClassVar[str] = "E3"
    title: ClassVar[str] = "Entity JSON Schema fails draft metaschema validation"

class UnresolvableJsonlSchemaRefError(JsonlJsonschemaError):
    """All `$ref` pointers and `allOf` branches in an entity JSON Schema must resolve without cycles before PyArrow conversion."""

    code: ClassVar[str] = "E4"
    title: ClassVar[str] = "Unresolvable or circular $ref in entity JSON Schema"

class UnsupportedJsonlArrowSchemaError(JsonlJsonschemaError):
    """Each non-transformed, non-overridden entity JSON Schema must have `type: "object"` with at least one property and only supported PyArrow column constructs."""

    code: ClassVar[str] = "E5"
    title: ClassVar[str] = "Entity JSON Schema cannot be converted to a PyArrow table schema"

class JsonlIngestSettings(FileFormatIngestSettings):
    """Settings for the JSONL ingest pipeline."""

    file_glob: Annotated[
        FileGlob,
        Field(
            default=DEFAULT_JSONL_FILE_GLOB,
            description=file_glob_description("Glob pattern for JSONL files"),
        ),
    ]

    def _resolve_entity_registry_schema_map(
        self: Any,
        registry: ModelRegistry | SchemaRegistry,
    ) -> None:
        """Validate configured tables against ``registry`` and resolve ``_schema_map``."""
        self._validate_configured_tables(
            registry.registry,
            validate_transforms_fn=validate_table_transforms,
        )
        self._schema_map = (
            build_output_schema_registry(
                registry,
                table_transforms=self._table_transforms,
                output_schemas=self._unreconciled_output_schemas,
            )
            if self.use_pyarrow
            else None
        )

class JsonlExtractSettings(JsonlIngestSettings, TransformSettingsMixin):
    """Settings for the JSONL extraction-only pipeline."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(
        cli_prog_name=EXTRACT_PIPELINE_NAME,
        cli_shortcuts=select_cli_shortcuts(LOCAL_CLI_SHORTCUTS, FILE_GLOB, TABLE_NAME),  # pyright: ignore[reportArgumentType]
    )

    table_name: TableName

class JsonlPydanticIngestSettings(JsonlIngestSettings, TransformSettingsMixin):
    """Pipeline settings for JSONL pipeline that uses Pydantic for validation."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(
        cli_prog_name=PYDANTIC_PIPELINE_NAME,
        cli_shortcuts=select_cli_shortcuts(LOCAL_CLI_SHORTCUTS, ENTITY_MODELS_MODULE, FILE_GLOB),  # pyright: ignore[reportArgumentType]
    )

    _model_registry: ModelRegistry = PrivateAttr()

    entity_models: Annotated[
        NonEmptyStr,
        Field(
            description=(
                "Dotted import path to a module that defines `ENTITY_MODELS: dict[str, type[BaseModel]]`. "
                "Each key is a table name and a subdirectory of `input_dir`. "
                "Each value is the Pydantic model for that table."
            ),
        ),
    ]

    def _resolve_settings_schema_map(self, default_pyarrow_schema: pa.Schema | None = None) -> None:  # noqa: ARG002
        """Load and validate the entity registry from the configured module and resolve `_schema_map`."""
        self._model_registry = load_registry_dict(self.entity_models, "ENTITY_MODELS")
        self._resolve_entity_registry_schema_map(self._model_registry)

    @property
    def model_registry(self) -> ModelRegistry:
        """The loaded `ENTITY_MODELS` mapping."""
        return self._model_registry

class JsonlJsonschemaIngestSettings(JsonlIngestSettings, TransformSettingsMixin):
    """Pipeline settings for JSONL ingestion with JSON Schema validation."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(
        cli_prog_name=JSONSCHEMA_PIPELINE_NAME,
        cli_shortcuts=select_cli_shortcuts(LOCAL_CLI_SHORTCUTS, FILE_GLOB, ENTITY_SCHEMAS_MODULE),  # pyright: ignore[reportArgumentType]
    )

    _schema_registry: SchemaRegistry = PrivateAttr()

    entity_schemas: Annotated[
        NonEmptyStr,
        Field(
            description=(
                "Dotted import path to a module that defines `ENTITY_SCHEMAS: dict[str, dict]`. "
                "Each key is a table name and a subdirectory of `input_dir`. "
                "Each value is a JSON Schema dictionary declaring its draft with `$schema`."
            ),
        ),
    ]

    def _resolve_settings_schema_map(self, default_pyarrow_schema: pa.Schema | None = None) -> None:  # noqa: ARG002
        """Load and validate the schema registry and precompute PyArrow schemas."""
        try:
            self._schema_registry = load_registry_dict(self.entity_schemas, "ENTITY_SCHEMAS")
        except Exception as exc:
            raise _classify_jsonl_schema_error(exc) from exc
        self._resolve_entity_registry_schema_map(self._schema_registry)

    @property
    def schema_registry(self) -> SchemaRegistry:
        """The loaded `ENTITY_SCHEMAS` mapping."""
        return self._schema_registry

class NcbiDatasetsSettings(JsonlJsonschemaIngestSettings):
    """Pipeline settings for NCBI Datasets JSONL ingestion (`annotation_report` and `dataset_report`)."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(
        cli_prog_name=NCBI_DATASETS_PIPELINE_NAME,
        cli_shortcuts=select_cli_shortcuts(LOCAL_CLI_SHORTCUTS, FILE_GLOB),  # pyright: ignore[reportArgumentType]
    )

    dataset_name: ClassVar[str] = NCBI_DATASETS_PIPELINE_NAME
    entity_schemas: ClassVar[str] = NCBI_DATASETS_SCHEMA_MODULE
    transforms_module: ClassVar[str | None] = NCBI_DATASETS_SCHEMA_MODULE
    output_schemas_module: ClassVar[str | None] = NCBI_DATASETS_SCHEMA_MODULE