Ormdantic¶

def table(
    self,
    tablename: str | None = None,
    *,
    pk: str,
    schema: str | None = None,
    indexed: list[str] | None = None,
    unique: list[str] | None = None,
    indexes: list[TableIndex] | None = None,
    column_options: dict[str, TableColumn] | None = None,
    unique_constraints: list[list[str] | TableUnique] | None = None,
    check_constraints: list[TableCheck] | None = None,
    foreign_key_constraints: list[TableForeignKey] | None = None,
    exclusion_constraints: list[TableExclusion] | None = None,
    comment: str | None = None,
    tablespace: str | None = None,
    mysql_engine: str | None = None,
    mysql_charset: str | None = None,
    mysql_collation: str | None = None,
    mysql_row_format: str | None = None,
    mysql_key_block_size: int | None = None,
    mysql_pack_keys: bool | None = None,
    mysql_checksum: bool | None = None,
    mysql_delay_key_write: bool | None = None,
    mysql_stats_persistent: bool | None = None,
    mysql_stats_auto_recalc: bool | None = None,
    mysql_stats_sample_pages: int | None = None,
    mysql_avg_row_length: int | None = None,
    mysql_max_rows: int | None = None,
    mysql_min_rows: int | None = None,
    mysql_insert_method: str | None = None,
    mysql_data_directory: str | None = None,
    mysql_index_directory: str | None = None,
    mysql_connection: str | None = None,
    mysql_union: list[str] | None = None,
    mysql_partition_by: str | None = None,
    mysql_partitions: int | None = None,
    mysql_subpartition_by: str | None = None,
    mysql_subpartitions: int | None = None,
    mysql_auto_increment: int | None = None,
    oracle_compress: int | bool | None = None,
    postgres_inherits: list[str] | None = None,
    postgres_with: dict[str, str | int | bool] | None = None,
    postgres_using: str | None = None,
    postgres_unlogged: bool = False,
    postgres_partition_by: str | None = None,
    postgres_partition_of: str | None = None,
    postgres_partition_for: str | None = None,
    sqlite_strict: bool = False,
    sqlite_without_rowid: bool = False,
    back_references: dict[str, str] | None = None,
) -> Callable[[Type[ModelType]], Type[ModelType]]:
    """Register a model as a database table."""

    def _wrapper(cls: Type[ModelType]) -> Type[ModelType]:
        """Decorator function."""
        tablename_ = tablename or snake_case(cls.__name__)
        schema_ = normalized_storage_identifier(
            schema,
            option_name=f"schema for table '{tablename_}'",
        )
        comment_ = normalized_table_comment(tablename_, comment)
        tablespace_ = normalized_storage_identifier(
            tablespace,
            option_name=f"tablespace for table '{tablename_}'",
        )
        mysql_engine_ = normalized_storage_token(
            mysql_engine,
            option_name=f"MySQL engine for table '{tablename_}'",
        )
        mysql_charset_ = normalized_storage_token(
            mysql_charset,
            option_name=f"MySQL charset for table '{tablename_}'",
        )
        mysql_collation_ = normalized_storage_token(
            mysql_collation,
            option_name=f"MySQL collation for table '{tablename_}'",
        )
        mysql_row_format_ = normalized_storage_token(
            mysql_row_format,
            option_name=f"MySQL row format for table '{tablename_}'",
        )
        mysql_key_block_size_ = normalized_positive_int(
            mysql_key_block_size,
            option_name=(f"MySQL/MariaDB KEY_BLOCK_SIZE for table '{tablename_}'"),
        )
        mysql_pack_keys_ = normalized_bool(
            mysql_pack_keys,
            option_name=f"MySQL/MariaDB PACK_KEYS for table '{tablename_}'",
        )
        mysql_checksum_ = normalized_bool(
            mysql_checksum,
            option_name=f"MySQL/MariaDB CHECKSUM for table '{tablename_}'",
        )
        mysql_delay_key_write_ = normalized_bool(
            mysql_delay_key_write,
            option_name=(f"MySQL/MariaDB DELAY_KEY_WRITE for table '{tablename_}'"),
        )
        mysql_stats_persistent_ = normalized_bool(
            mysql_stats_persistent,
            option_name=(
                f"MySQL/MariaDB STATS_PERSISTENT for table '{tablename_}'"
            ),
        )
        mysql_stats_auto_recalc_ = normalized_bool(
            mysql_stats_auto_recalc,
            option_name=(
                f"MySQL/MariaDB STATS_AUTO_RECALC for table '{tablename_}'"
            ),
        )
        mysql_stats_sample_pages_ = normalized_positive_int(
            mysql_stats_sample_pages,
            option_name=(
                f"MySQL/MariaDB STATS_SAMPLE_PAGES for table '{tablename_}'"
            ),
        )
        mysql_avg_row_length_ = normalized_positive_int(
            mysql_avg_row_length,
            option_name=f"MySQL/MariaDB AVG_ROW_LENGTH for table '{tablename_}'",
        )
        mysql_max_rows_ = normalized_positive_int(
            mysql_max_rows,
            option_name=f"MySQL/MariaDB MAX_ROWS for table '{tablename_}'",
        )
        mysql_min_rows_ = normalized_positive_int(
            mysql_min_rows,
            option_name=f"MySQL/MariaDB MIN_ROWS for table '{tablename_}'",
        )
        mysql_insert_method_ = normalized_storage_token(
            mysql_insert_method,
            option_name=f"MySQL/MariaDB INSERT_METHOD for table '{tablename_}'",
        )
        mysql_data_directory_ = normalized_storage_path(
            mysql_data_directory,
            option_name=f"MySQL/MariaDB DATA DIRECTORY for table '{tablename_}'",
        )
        mysql_index_directory_ = normalized_storage_path(
            mysql_index_directory,
            option_name=f"MySQL/MariaDB INDEX DIRECTORY for table '{tablename_}'",
        )
        mysql_connection_ = normalized_storage_string(
            mysql_connection,
            option_name=f"MySQL/MariaDB CONNECTION for table '{tablename_}'",
        )
        mysql_union_ = normalized_storage_identifier_list(
            mysql_union,
            option_name=f"MySQL/MariaDB UNION for table '{tablename_}'",
        )
        mysql_partition_by_ = normalized_mysql_partition_by(
            mysql_partition_by,
            table_name=tablename_,
        )
        mysql_partitions_ = normalized_positive_int(
            mysql_partitions,
            option_name=f"MySQL/MariaDB PARTITIONS for table '{tablename_}'",
        )
        mysql_subpartition_by_ = normalized_mysql_partition_by(
            mysql_subpartition_by,
            table_name=tablename_,
            option="SUBPARTITION BY",
        )
        mysql_subpartitions_ = normalized_positive_int(
            mysql_subpartitions,
            option_name=f"MySQL/MariaDB SUBPARTITIONS for table '{tablename_}'",
        )
        mysql_auto_increment_ = normalized_positive_int(
            mysql_auto_increment,
            option_name=f"MySQL/MariaDB AUTO_INCREMENT for table '{tablename_}'",
        )
        oracle_compress_ = normalized_oracle_table_compress(
            oracle_compress,
            table_name=tablename_,
        )
        postgres_inherits_ = [
            normalized_storage_identifier(
                parent,
                option_name=f"PostgreSQL inherited table for table '{tablename_}'",
            )
            or ""
            for parent in postgres_inherits or []
        ]
        postgres_with_ = normalized_postgres_storage_parameters(
            postgres_with,
            table_name=tablename_,
        )
        postgres_using_ = normalized_storage_token(
            postgres_using,
            option_name=f"PostgreSQL table access method for table '{tablename_}'",
        )
        postgres_partition_by_ = normalized_postgres_partition_by(
            postgres_partition_by,
            table_name=tablename_,
        )
        postgres_partition_of_ = normalized_storage_identifier(
            postgres_partition_of,
            option_name=f"PostgreSQL partition parent for table '{tablename_}'",
        )
        postgres_partition_for_ = normalized_postgres_partition_for(
            postgres_partition_for,
            table_name=tablename_,
        )
        if (postgres_partition_of_ is None) != (postgres_partition_for_ is None):
            raise ValueError(
                f"PostgreSQL partition table '{tablename_}' requires both "
                "postgres_partition_of and postgres_partition_for"
            )
        if postgres_partition_of_ is not None and postgres_inherits_:
            raise ValueError(
                f"PostgreSQL partition table '{tablename_}' cannot also use "
                "postgres_inherits"
            )
        indexes_ = indexes or []
        anonymous_unique_constraints, named_unique_constraints = (
            split_unique_constraints(unique_constraints or [])
        )
        clustered_names = [
            f"index {index.name}" for index in indexes_ if index.mssql_clustered
        ]
        clustered_names.extend(
            f"unique constraint {constraint.name}"
            for constraint in named_unique_constraints
            if constraint.mssql_clustered is True
        )
        if len(clustered_names) > 1:
            names = ", ".join(clustered_names)
            raise ValueError(
                f"table '{tablename_}' has multiple SQL Server clustered "
                f"indexes or unique constraints: {names}"
            )
        cls_back_references = back_references or {}
        fields = model_fields(cls)
        column_options_ = column_options or {}
        unknown_options = set(column_options_) - set(fields)
        if unknown_options:
            unknown = ", ".join(sorted(unknown_options))
            raise ValueError(
                f"column_options for table '{tablename_}' reference unknown fields: {unknown}"
            )
        foreign_key_constraints_ = foreign_key_constraints or []
        unknown_foreign_key_columns = sorted(
            {
                column
                for constraint in foreign_key_constraints_
                for column in constraint.columns
                if column not in fields
            }
        )
        if unknown_foreign_key_columns:
            unknown = ", ".join(unknown_foreign_key_columns)
            raise ValueError(
                f"foreign_key_constraints for table '{tablename_}' "
                f"reference unknown fields: {unknown}"
            )
        exclusion_constraints_ = exclusion_constraints or []
        unknown_exclusion_columns = sorted(
            {
                column
                for constraint in exclusion_constraints_
                for column, _operator in constraint.columns
                if column not in fields
            }
        )
        if unknown_exclusion_columns:
            unknown = ", ".join(unknown_exclusion_columns)
            raise ValueError(
                f"exclusion_constraints for table '{tablename_}' "
                f"reference unknown fields: {unknown}"
            )
        table_metadata = OrmTable[ModelType](
            model=cls,
            tablename=tablename_,
            pk=pk,
            schema_name=schema_,
            comment=comment_,
            tablespace=tablespace_,
            mysql_engine=mysql_engine_,
            mysql_charset=mysql_charset_,
            mysql_collation=mysql_collation_,
            mysql_row_format=mysql_row_format_,
            mysql_key_block_size=mysql_key_block_size_,
            mysql_pack_keys=mysql_pack_keys_,
            mysql_checksum=mysql_checksum_,
            mysql_delay_key_write=mysql_delay_key_write_,
            mysql_stats_persistent=mysql_stats_persistent_,
            mysql_stats_auto_recalc=mysql_stats_auto_recalc_,
            mysql_stats_sample_pages=mysql_stats_sample_pages_,
            mysql_avg_row_length=mysql_avg_row_length_,
            mysql_max_rows=mysql_max_rows_,
            mysql_min_rows=mysql_min_rows_,
            mysql_insert_method=mysql_insert_method_,
            mysql_data_directory=mysql_data_directory_,
            mysql_index_directory=mysql_index_directory_,
            mysql_connection=mysql_connection_,
            mysql_union=mysql_union_,
            mysql_partition_by=mysql_partition_by_,
            mysql_partitions=mysql_partitions_,
            mysql_subpartition_by=mysql_subpartition_by_,
            mysql_subpartitions=mysql_subpartitions_,
            mysql_auto_increment=mysql_auto_increment_,
            oracle_compress=oracle_compress_,
            postgres_inherits=postgres_inherits_,
            postgres_with=postgres_with_,
            postgres_using=postgres_using_,
            postgres_unlogged=postgres_unlogged,
            postgres_partition_by=postgres_partition_by_,
            postgres_partition_of=postgres_partition_of_,
            postgres_partition_for=postgres_partition_for_,
            sqlite_strict=sqlite_strict,
            sqlite_without_rowid=sqlite_without_rowid,
            indexed=indexed or [],
            unique=unique or [],
            indexes=indexes_,
            column_options=column_options_,
            unique_constraints=anonymous_unique_constraints,
            named_unique_constraints=named_unique_constraints,
            check_constraints=check_constraints or [],
            foreign_key_constraints=foreign_key_constraints_,
            exclusion_constraints=exclusion_constraints_,
            columns=[field for field in fields if field not in cls_back_references],
            relationships={},
            back_references=cls_back_references,
        )
        self._table_map.model_to_data[cls] = table_metadata
        self._table_map.name_to_data[tablename_] = table_metadata
        return cls

    return _wrapper

def namespace(self, name: str, *, comment: str | None = None) -> DatabaseNamespace:
    """Register a database namespace/schema for snapshots and migrations."""
    namespace = DatabaseNamespace(name=name, comment=comment)
    if any(existing.name == namespace.name for existing in self._namespaces):
        raise ValueError(f"duplicate namespace '{namespace.name}'")
    self._namespaces.append(namespace)
    return namespace

def sequence(
    self,
    name: str,
    *,
    schema: str | None = None,
    data_type: str | None = None,
    start: int | None = None,
    increment: int | None = None,
    min_value: int | None = None,
    max_value: int | None = None,
    no_min_value: bool = False,
    no_max_value: bool = False,
    cycle: bool = False,
    cache: int | None = None,
    comment: str | None = None,
    order: bool = False,
) -> DatabaseSequence:
    """Register a database sequence for schema snapshots and migrations."""
    sequence = DatabaseSequence(
        name=name,
        schema=schema,
        data_type=data_type,
        start=start,
        increment=increment,
        min_value=min_value,
        max_value=max_value,
        no_min_value=no_min_value,
        no_max_value=no_max_value,
        cycle=cycle,
        cache=cache,
        comment=comment,
        order=order,
    )
    key = (sequence.schema_name, sequence.name)
    if any(
        (existing.schema_name, existing.name) == key for existing in self._sequences
    ):
        qualified = (
            sequence.name
            if sequence.schema_name is None
            else f"{sequence.schema_name}.{sequence.name}"
        )
        raise ValueError(f"duplicate sequence '{qualified}'")
    self._sequences.append(sequence)
    return sequence

def view(
    self,
    name: str,
    definition: str,
    *,
    schema: str | None = None,
    materialized: bool = False,
    comment: str | None = None,
) -> DatabaseView:
    """Register a database view for schema snapshots and migrations."""
    view = DatabaseView(
        name=name,
        schema=schema,
        definition=definition,
        materialized=materialized,
        comment=comment,
    )
    key = (view.schema_name, view.name)
    if any(
        (existing.schema_name, existing.name) == key for existing in self._views
    ):
        qualified = (
            view.name
            if view.schema_name is None
            else f"{view.schema_name}.{view.name}"
        )
        raise ValueError(f"duplicate view '{qualified}'")
    self._views.append(view)
    return view

async def init(self) -> None:
    """Initialize ORM models."""
    for table_data in self._table_map.name_to_data.values():
        rels = self.get(table_data)
        table_data.relationships = rels
    for table_data in self._table_map.name_to_data.values():
        for field_name in table_data.relationships:
            install_relationship_path_descriptor(table_data.model, field_name)
    self._runtime = self._build_runtime_database()
    for table_data in self._table_map.name_to_data.values():
        self._tables[table_data.model] = Table(
            table_data=table_data,
            table_map=self._table_map,
            rust_handle=self._runtime.table(table_data.model.__name__),
            events=self._events,
            runtime=self._runtime,
        )
    await self.create_all()

async def create_all(self) -> None:
    """Create all registered tables."""
    if self._runtime is None:
        self._runtime = self._build_runtime_database()
    self._create_registered_namespaces()
    self._create_registered_sequences()
    self._runtime.create_all()
    self._create_registered_postgres_unique_options()
    self._create_registered_constraint_comments()
    self._create_registered_postgres_index_options()
    self._create_registered_mssql_index_options()
    self._create_registered_oracle_index_options()
    self._create_registered_mysql_index_options()
    self._create_registered_index_tablespaces()
    self._create_registered_index_comments()
    self._create_registered_enum_type_comments()
    self._create_registered_views()

async def drop_all(self) -> None:
    """Drop all registered tables."""
    if self._runtime is not None:
        self._drop_registered_views()
        self._runtime.drop_all()
        self._drop_registered_sequences()
        self._drop_registered_namespaces()
        return
    self._drop_registered_views()
    for table_data in reversed(list(self._table_map.name_to_data.values())):
        sql = _ormdantic.compile_drop_table_sql(
            self._connection, table_data.tablename
        )
        _ormdantic.execute_native(self._connection, sql, [])
    for enum_type in reversed(self._runtime_enum_type_specs()):
        _ormdantic.execute_native(
            self._connection,
            _drop_runtime_enum_type_sql(enum_type),
            [],
        )
    self._drop_registered_sequences()
    self._drop_registered_namespaces()

def transaction(
    self,
    *,
    isolation_level: TransactionIsolationLevel | str | None = None,
    read_only: bool = False,
    deferrable: bool | None = None,
) -> Any:
    """Open a native transaction context."""
    return _OrmdanticTransaction(
        self,
        self._transaction_options(
            isolation_level=isolation_level,
            read_only=read_only,
            deferrable=deferrable,
        ),
    )

def session(
    self,
    *,
    isolation_level: TransactionIsolationLevel | str | None = None,
    read_only: bool = False,
    deferrable: bool | None = None,
) -> Session:
    """Open an async unit-of-work session."""
    return Session(
        self,
        transaction_options=self._transaction_options(
            isolation_level=isolation_level,
            read_only=read_only,
            deferrable=deferrable,
        ),
    )

def inspect(self) -> Inspector:
    """Return a database inspector."""
    return Inspector(self)

def relation(
    self,
    model: Type[ModelType],
    relationship: str,
    *,
    outer_alias: str | None = None,
    target_alias: str | None = None,
) -> RelationExpression:
    """Return a typed helper for relationship predicates and aggregates."""
    table = self._table_map.model_to_data.get(model)
    if table is not None and not table.relationships:
        table.relationships = self.get(table)
    return relation_expression(
        self._table_map,
        model,
        relationship,
        outer_alias=outer_alias,
        target_alias=target_alias,
    )

def savepoint(self, name: str) -> Any:
    """Open a savepoint context."""
    return _OrmdanticSavepoint(self, name)

def on(self, event: str, handler: EventHandler) -> EventHandler:
    """Register an event handler."""
    return self._events.on(event, handler)

def off(self, event: str, handler: EventHandler) -> None:
    """Remove a registered event handler."""
    self._events.off(event, handler)

def clear_events(self, event: str | None = None) -> None:
    """Clear event handlers for one event or all events."""
    self._events.clear(event)

async def load(self, model: ModelType, path: LoaderPathLike) -> Any:
    """Explicitly load a relationship path for a model instance."""
    table = self._table_map.model_to_data[type(model)]
    normalized_path = ".".join(path_parts(path))
    loaded = await self[type(model)].find_one(
        getattr(model, table.pk), load=[joinedload(normalized_path)]
    )
    if loaded is None:
        return None
    value: Any = loaded
    for part in path_parts(normalized_path):
        value = getattr(value, part)
    return value

def get(self, table_data: OrmTable[ModelType]) -> dict[str, Relationship]:
    """Get relationships for a given table."""
    relationships = {}
    for field_name, field in model_fields(table_data.model).items():
        related_table = self._get_related_table(field)
        if related_table is None:
            continue
        if back_reference := table_data.back_references.get(field_name):
            relationships[field_name] = self._get_many_relationship(
                field_name, back_reference, table_data, related_table
            )

            continue
        if contains_list_annotation(
            field.annotation
        ) or field.annotation == ForwardRef(f"{related_table.model.__name__}"):
            raise UndefinedBackReferenceError(
                table_data.tablename, related_table.tablename, field_name
            )

        args = get_args(field.annotation)
        correct_type = (
            model_field(related_table.model, related_table.pk).annotation in args
        )
        origin = get_origin(field.annotation)
        if not args or origin not in {UnionType, Union} or not correct_type:
            raise MustUnionForeignKeyError(
                table_data.tablename,
                related_table.tablename,
                field_name,
                related_table.model,
                model_field(
                    related_table.model, related_table.pk
                ).annotation.__name__,
            )

        relationships[field_name] = Relationship(
            foreign_table=related_table.tablename
        )

    return relationships