biodm.components.table module

SQLAlchemy tables declarative parent classes - Declarative Base - S3File entity - Versioned

class Base(**kwargs: Any)

Bases: DeclarativeBase, AsyncAttrs

Base class for ORM declarative Tables.

Parameters:

  • svc (DatabaseService) – Enable entity - service linkage

  • ctrl (ResourceController) – Enable entity - controller linkage -> Resources tables only

  • is_permission (bool) – Is permission table flag

  • last_max_id (int) – Support non duplicate sqlite id population

async clone(session: AsyncSession, new_item: Self | None = None, updated_rels: List[str] | None = None) → Self

Deep clone an entity with its relationships.

Parameters:

  • session (AsyncSession) – session

  • new_item (Self | None, optional) – New instance, if None -> generated with make_transient, defaults to None

  • updated_rels (List[str] | None, optional) – list of relationship to ignore while cloning, defaults to None

Raises:

NotImplementedError – canary, for versioned case of one-to-one

Returns:

cloned item

Return type:

Self

classmethod col(name: str)

Return columns object from name.

classmethod colinfo(name: str) → Tuple[Column, type]

Return column and associated python type for conditions.

ctrl: ClassVar[Type[ResourceController]]
classmethod dyn_relationships()

Return table relationships. dyn stands for dynamic -> use for setup.

has_composite_pk_with_leading_id_sqlite = False
classmethod has_default(name: str) → bool

Flag if column has default value.

has_submitter_username = False
classmethod is_autoincrement(name: str) → bool

Flag if column is autoincrement.

Warning! This check is backend dependent and should be changed when supporting a new one. E.g. Oracle backend will not react appropriately. - https://groups.google.com/g/sqlalchemy/c/o5YQNH5UUko

is_permission: bool = False
is_versioned = False
last_max_id: int = 0
metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

pk() → OrderedSet[str]

Return primary key names.

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

relationships()

Table relationships. Memoized result -> use for runtime.

required() → Set[str]

Gets all required fields to create a new entry in this table.

Returns:

fields name list

Return type:

Set[str]

svc: ClassVar[Type[DatabaseService]]
classmethod target_table(name)

Return target table of a property.

class S3File

Bases: object

Class to use in order to have a file managed on S3 bucket associated to this table Defaults internal fields that are expected by S3Service.

dl_count = Column(None, Integer(), table=None, nullable=False, server_default=DefaultClause('0', for_update=False))
emited_at = Column(None, TIMESTAMP(timezone=True), table=None, nullable=False, default=CallableColumnDefault(<function utcnow>))
extension = Column(None, String(length=10), table=None, nullable=False)
filename = Column(None, String(length=100), table=None, nullable=False)
key
key_salt = Column(None, String(), table=None, nullable=False, default=CallableColumnDefault(<function S3File.<lambda>>))
ready = Column(None, BOOLEAN(), table=None, nullable=False, server_default=DefaultClause('0', for_update=False))
size = Column(None, BigInteger(), table=None, nullable=False)
upload = <_RelationshipDeclared at 0x7fbdae997cf0; no key>
upload_id: Mapped[int] = <sqlalchemy.orm.properties.MappedColumn object>
validated_at = Column(None, TIMESTAMP(timezone=True), table=None)
class Versioned

Bases: object

Versioned entity parent class.

  • Populates version as primary_key column

  • Disable /update, enable /release

version = Column(None, Integer(), table=None, primary_key=True, nullable=False, server_default=DefaultClause('1', for_update=False))
add_versioned_table_methods() → None

Called after tables initialization to have access to aliased which is not the case during initialization.