lsst.dax.apdb.sql.modelToSql.ModelToSql Class Reference

Public Member Functions
	__init__ (self, sqlalchemy.schema.MetaData metadata, str prefix="")
Mapping[str, sqlalchemy.schema.Table]	make_tables (self, Iterable[schema_model.Table] tables)

Protected Member Functions
list[sqlalchemy.schema.Column]	_table_columns (self, schema_model.Table table)
list[sqlalchemy.schema.SchemaItem]	_table_constraints (self, schema_model.Table table, Mapping[str, sqlalchemy.schema.Table] table_map)

Static Protected Member Functions
list[schema_model.Table]	_topo_sort (Iterable[schema_model.Table] table_iter)

Protected Attributes
	_metadata
	_prefix

Detailed Description

Class which implements schema model conversion to SQLAlchemy format.

Parameters
----------
metadata : `sqlalchemy.schema.MetaData`
    Metadata object for created tables.
prefix : `str`, optional
    Prefix to add to all schema elements.

Definition at line 90 of file modelToSql.py.

Constructor & Destructor Documentation

__init__()

lsst.dax.apdb.sql.modelToSql.ModelToSql.__init__	(		self,
		sqlalchemy.schema.MetaData	metadata,
		str	prefix = "" )

Definition at line 101 of file modelToSql.py.

    ):
        self._metadata = metadata
        self._prefix = prefix
 
        # Map model column types to SQLAlchemy.
        self._type_map: dict[felis.datamodel.DataType | schema_model.ExtraDataTypes, type] = {
            felis.datamodel.DataType.double: sqlalchemy.types.Double,
            felis.datamodel.DataType.float: sqlalchemy.types.Float,
            felis.datamodel.DataType.timestamp: sqlalchemy.types.TIMESTAMP,
            felis.datamodel.DataType.long: sqlalchemy.types.BigInteger,
            felis.datamodel.DataType.int: sqlalchemy.types.Integer,
            felis.datamodel.DataType.short: sqlalchemy.types.SmallInteger,
            felis.datamodel.DataType.byte: sqlalchemy.types.SmallInteger,  # Byte types are not very portable
            felis.datamodel.DataType.binary: sqlalchemy.types.LargeBinary,
            felis.datamodel.DataType.text: sqlalchemy.types.Text,
            felis.datamodel.DataType.string: sqlalchemy.types.CHAR,
            felis.datamodel.DataType.char: sqlalchemy.types.CHAR,
            felis.datamodel.DataType.unicode: sqlalchemy.types.CHAR,
            felis.datamodel.DataType.boolean: sqlalchemy.types.Boolean,
            schema_model.ExtraDataTypes.UUID: GUID,
        }
 

Member Function Documentation

_table_columns()

list[sqlalchemy.schema.Column] lsst.dax.apdb.sql.modelToSql.ModelToSql._table_columns ( self, schema_model.Table table )

protected

Return set of columns in a table

Parameters
----------
table : `schema_model.Table`
    Table model.

Returns
-------
column_defs : `list` [`sqlalchemy.schema.Column`]
    List of columns.

Definition at line 159 of file modelToSql.py.

    def _table_columns(self, table: schema_model.Table) -> list[sqlalchemy.schema.Column]:
        """Return set of columns in a table
 
        Parameters
        ----------
        table : `schema_model.Table`
            Table model.
 
        Returns
        -------
        column_defs : `list` [`sqlalchemy.schema.Column`]
            List of columns.
        """
        column_defs: list[sqlalchemy.schema.Column] = []
        for column in table.columns:
            kwargs: dict[str, Any] = dict(nullable=column.nullable)
            if column.value is not None:
                kwargs.update(server_default=str(column.value))
            if column in table.primary_key and column.autoincrement is None:
                kwargs.update(autoincrement=False)
            else:
                kwargs.update(autoincrement=column.autoincrement)
            ctype = self._type_map[column.datatype]
            if column.length is not None:
                if ctype not in (sqlalchemy.types.Text, sqlalchemy.types.TIMESTAMP):
                    ctype = ctype(length=column.length)
            column_defs.append(sqlalchemy.schema.Column(column.name, ctype, **kwargs))
 
        return column_defs
 

_table_constraints()

list[sqlalchemy.schema.SchemaItem] lsst.dax.apdb.sql.modelToSql.ModelToSql._table_constraints ( self, schema_model.Table table, Mapping[str, sqlalchemy.schema.Table] table_map )

protected

Return set of constraints/indices in a table.

Parameters
----------
table : `schema_model.Table`
    Table model.
table_map : `~collections.abc.Mapping`
    MApping of table ID to sqlalchemy table definition for tables
    that already exist, this must include all tables referenced by
    foreign keys in ``table``.

Returns
-------
constraints : `list` [`sqlalchemy.schema.SchemaItem`]
    List of SQLAlchemy index/constraint objects.

Definition at line 189 of file modelToSql.py.

    ) -> list[sqlalchemy.schema.SchemaItem]:
        """Return set of constraints/indices in a table.
 
        Parameters
        ----------
        table : `schema_model.Table`
            Table model.
        table_map : `~collections.abc.Mapping`
            MApping of table ID to sqlalchemy table definition for tables
            that already exist, this must include all tables referenced by
            foreign keys in ``table``.
 
        Returns
        -------
        constraints : `list` [`sqlalchemy.schema.SchemaItem`]
            List of SQLAlchemy index/constraint objects.
        """
        constraints: list[sqlalchemy.schema.SchemaItem] = []
        if table.primary_key:
            # It is very useful to have named PK.
            name = self._prefix + table.name + "_pk"
            constraints.append(
                sqlalchemy.schema.PrimaryKeyConstraint(*[column.name for column in table.primary_key])
            )
        for index in table.indexes:
            if index.expressions:
                raise TypeError(f"Expression indices are not supported: {table}")
            name = self._prefix + index.name if index.name else ""
            constraints.append(sqlalchemy.schema.Index(name, *[column.name for column in index.columns]))
        for constraint in table.constraints:
            constr_name: str | None = None
            if constraint.name:
                constr_name = self._prefix + constraint.name
            if isinstance(constraint, schema_model.UniqueConstraint):
                constraints.append(
                    sqlalchemy.schema.UniqueConstraint(
                        *[column.name for column in constraint.columns], name=constr_name
                    )
                )
            elif isinstance(constraint, schema_model.ForeignKeyConstraint):
                column_names = [col.name for col in constraint.columns]
                foreign_table = table_map[constraint.referenced_table.id]
                refcolumns = [foreign_table.columns[col.name] for col in constraint.referenced_columns]
                constraints.append(
                    sqlalchemy.schema.ForeignKeyConstraint(
                        columns=column_names,
                        refcolumns=refcolumns,
                        name=constr_name,
                        deferrable=constraint.deferrable,
                        initially=constraint.initially,
                        onupdate=constraint.onupdate,
                        ondelete=constraint.ondelete,
                    )
                )
            elif isinstance(constraint, schema_model.CheckConstraint):
                constraints.append(
                    sqlalchemy.schema.CheckConstraint(
                        constraint.expression,
                        name=constr_name,
                        deferrable=constraint.deferrable,
                        initially=constraint.initially,
                    )
                )
            else:
                raise TypeError(f"Unknown constraint type: {constraint}")
 
        return constraints
 

_topo_sort()

list[schema_model.Table] lsst.dax.apdb.sql.modelToSql.ModelToSql._topo_sort ( Iterable[schema_model.Table] table_iter )

staticprotected

Toplogical sorting of tables.

Definition at line 262 of file modelToSql.py.

    def _topo_sort(table_iter: Iterable[schema_model.Table]) -> list[schema_model.Table]:
        """Toplogical sorting of tables."""
        result: list[schema_model.Table] = []
        result_ids: set[str] = set()
        tables = list(table_iter)
 
        # Map of table ID to foreign table IDs.
        referenced_tables: dict[str, set[str]] = {}
        for table in tables:
            referenced_tables[table.id] = set()
            for constraint in table.constraints:
                if isinstance(constraint, schema_model.ForeignKeyConstraint):
                    referenced_tables[table.id].add(constraint.referenced_table.id)
 
        while True:
            keep = []
            changed = False
            for table in tables:
                if referenced_tables[table.id].issubset(result_ids):
                    changed = True
                    result.append(table)
                    result_ids.add(table.id)
                else:
                    keep.append(table)
            tables = keep
            if not changed:
                break
 
        # If nothing can be removed it means cycle.
        if tables:
            raise ValueError(f"Dependency cycle in foreign keys: {tables}")
 
        return result

make_tables()

Mapping[str, sqlalchemy.schema.Table] lsst.dax.apdb.sql.modelToSql.ModelToSql.make_tables	(		self,
		Iterable[schema_model.Table]	tables )

Generate sqlalchemy table schema from the list of modedls.

Parameters
----------
tables : `~collections.abc.Iterable` [`schema_model.Table`]
    List of table models.

Returns
-------
tables : `~collections.abc.Mapping` [`str`, `sqlalchemy.schema.Table`]
    SQLAlchemy table definitions indexed by identifier of the table
    model.

Definition at line 127 of file modelToSql.py.

    def make_tables(self, tables: Iterable[schema_model.Table]) -> Mapping[str, sqlalchemy.schema.Table]:
        """Generate sqlalchemy table schema from the list of modedls.
 
        Parameters
        ----------
        tables : `~collections.abc.Iterable` [`schema_model.Table`]
            List of table models.
 
        Returns
        -------
        tables : `~collections.abc.Mapping` [`str`, `sqlalchemy.schema.Table`]
            SQLAlchemy table definitions indexed by identifier of the table
            model.
        """
        # Order tables based on their FK dependencies.
        tables = self._topo_sort(tables)
 
        table_map: dict[str, sqlalchemy.schema.Table] = {}
        for table in tables:
            columns = self._table_columns(table)
            constraints = self._table_constraints(table, table_map)
            sa_table = sqlalchemy.schema.Table(
                self._prefix + table.name,
                self._metadata,
                *columns,
                *constraints,
                schema=self._metadata.schema,
            )
            table_map[table.id] = sa_table
 
        return table_map
 

Member Data Documentation

_metadata

lsst.dax.apdb.sql.modelToSql.ModelToSql._metadata

protected

Definition at line 106 of file modelToSql.py.

_prefix

lsst.dax.apdb.sql.modelToSql.ModelToSql._prefix

protected

Definition at line 107 of file modelToSql.py.

---

The documentation for this class was generated from the following file:

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-8.0.0/Linux64/dax_apdb/g1afd7665f7+eb25d4c773/python/lsst/dax/apdb/sql/modelToSql.py