Module madsci.common.types.document_db_migration_types

Document database migration configuration types.

Classes¶

CollectionDefinition(**data: Any)

Document database collection definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)¶

• pydantic.main.BaseModel

Class variables¶

description: str | None :

indexes: List[madsci.common.types.document_db_migration_types.IndexDefinition] :

model_config :

Static methods¶

validate_indexes(v: Any) ‑> List[madsci.common.types.document_db_migration_types.IndexDefinition]

Validate and convert indexes from various formats.

Methods¶

to_schema_dict(self) ‑> Dict[str, Any]

Convert to schema.json format.

DocumentDBMigrationSettings(**kwargs: Any)

Configuration settings for document database migration operations.

Initialize settings with walk-up file discovery.

Configuration file paths (YAML, JSON, TOML, .env) are resolved via walk-up discovery from a starting directory. Each filename walks up independently, so node.settings.yaml can resolve in the node dir while settings.yaml resolves in the lab root.

The starting directory is determined by (in priority order):

1. _settings_dir keyword argument

2. MADSCI_SETTINGS_DIR environment variable

3. Current working directory (default)

Args: _settings_dir: Starting directory for walk-up file discovery. **kwargs: Forwarded to BaseSettings.__init__.

Ancestors (in MRO)¶

• madsci.common.types.base_types.MadsciBaseSettings

• pydantic_settings.main.BaseSettings

• pydantic.main.BaseModel

Class variables¶

backup_dir: str | pathlib.Path :

backup_only: bool :

check_version: bool :

database: str | None :

document_db_url: pydantic.networks.AnyUrl :

restore_from: str | pathlib.Path | None :

schema_file: str | pathlib.Path | None :

target_version: str | None :

validate_schema: bool :

Methods¶

get_effective_schema_file_path(self) ‑> pathlib.Path

Get the effective schema file path as a Path object.

IndexDefinition(**data: Any)

Document database index definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)¶

• pydantic.main.BaseModel

Class variables¶

background: bool :

description: str | None :

keys: List[madsci.common.types.document_db_migration_types.IndexKey] :

model_config :

name: str :

unique: bool :

Static methods¶

validate_keys(v: Any) ‑> List[madsci.common.types.document_db_migration_types.IndexKey]

Validate and convert keys from various formats.

Methods¶

get_keys_as_tuples(self) ‑> List[tuple[str, int]]

Get index keys as tuples for MongoDB operations.

to_mongo_format(self) ‑> Dict[str, Any]

Convert to MongoDB index creation format.

to_schema_dict(self) ‑> Dict[str, Any]

Convert to schema.json format.

IndexKey(**data: Any)

Represents a single key in a document database index.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)¶

• pydantic.main.BaseModel

Class variables¶

direction: int :

field: str :

model_config :

Static methods¶

from_list(key_list: List[Any]) ‑> madsci.common.types.document_db_migration_types.IndexKey

Create IndexKey from list format [field, direction].

validate_direction(v: int) ‑> int

Validate index direction is 1 or -1.

Methods¶

to_list(self) ‑> list

Convert to list format for schema.json.

to_tuple(self) ‑> tuple[str, int]

Convert to tuple format for MongoDB operations.

MongoDBSchema(**data: Any)

Complete MongoDB database schema definition using Pydantic models

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)¶

• pydantic.main.BaseModel

Class variables¶

collections: Dict[str, madsci.common.types.document_db_migration_types.CollectionDefinition] :

database: str :

description: str | None :

model_config :

schema_version: pydantic_extra_types.semantic_version.SemanticVersion :

Static methods¶

from_file(file_path: str) ‑> madsci.common.types.document_db_migration_types.MongoDBSchema

Load schema from a JSON file.

from_mongodb_database(database_name: str, mongo_client: Any, schema_version: str = '0.0.0') ‑> madsci.common.types.document_db_migration_types.MongoDBSchema

Extract schema from an existing MongoDB database.

Args: database_name: Name of the database mongo_client: PyMongo MongoClient instance schema_version: Version to assign to the extracted schema

Returns: MongoDBSchema instance representing the database’s current schema

validate_collections(v: Any) ‑> Dict[str, madsci.common.types.document_db_migration_types.CollectionDefinition]

Validate and convert collections from various formats.

validate_version(v: Any) ‑> pydantic_extra_types.semantic_version.SemanticVersion

Validate and parse semantic version.

Methods¶

compare_with_database_schema(self, db_schema: MongoDBSchema) ‑> Dict[str, Any]

Compare this schema with a database schema and return differences.

Returns: Dictionary with differences: - missing_collections: Collections in expected schema but not in DB - extra_collections: Collections in DB but not in expected schema - collection_differences: Per-collection index differences

get_collection(self, collection_name: str) ‑> madsci.common.types.document_db_migration_types.CollectionDefinition | None

Get collection definition by name.

get_collection_names(self) ‑> List[str]

Get list of collection names.

has_collection(self, collection_name: str) ‑> bool

Check if collection exists in schema.

to_file(self, file_path: str, indent: int = 2) ‑> None

Save schema to a JSON file.

to_schema_dict(self) ‑> Dict[str, Any]

Convert to schema.json format.