from __future__ import annotations
import typing
from typing import ClassVar
import types
import re
import sys
import inspect
import itertools
import asyncio
if sys.version_info >= (3, 14):
import annotationlib
try:
from pydantic import TypeAdapter, ConfigDict
VALIDATE_DATA = True
PYDANTIC_CONFIG = ConfigDict(arbitrary_types_allowed=True, strict=False)
except ImportError:
VALIDATE_DATA = False
pass
from .db_field import DBField, UX, DBManyField, DBManyToManyField
from .db_types import *
from .db_types import DBTableT
from .exceptions import *
if typing.TYPE_CHECKING:
from typing import *
from .db_factory import DBFactory, DBFactoryAsync
from .db_query import DBQuery, DBQueryAsync, DBQueryField, DBSQL, FDBSQL
__all__ = ['DBTable', 'dbtable']
def camel2snake(name: str) -> str:
return camel2snake.r.sub(r'_\1', name).lower()
camel2snake.r = re.compile(
'((?<=[a-z0-9])[A-Z]|(?!^)(?<!_)[A-Z](?=[a-z]))') # tnx to https://stackoverflow.com/questions/1175208/elegant-python-function-to-convert-camelcase-to-snake-case
def get_annotated_id(t: str) -> str | None:
groups = get_annotated_id.r.match(t)
return groups and groups.group(2)
get_annotated_id.r = re.compile(r'''^(.*?)(Many|ManyToMany|Property|FieldCID|ClassVar|ObjVar)\[(.+?)]$''')
class MetaTable(type):
db_base_class: type[DBTable.DB]
def __new__(cls, clsname: str, bases: tuple[type[DBTable], ...], attrs: dict[str, Any], **kwargs: Any) -> type[DBTable]:
if clsname == 'DBTable':
cls.db_base_class = attrs['DB']
return super().__new__(cls, clsname, bases, attrs)
if 'DB' in attrs:
raise QuazyError(f'Should not define `DB` subclass directly in `{clsname}`, use `_name_` attributes')
spec_attrs = {}
for name in ('db table title schema just_for_typing extendable discriminator'
' meta lookup_field validate metadata use_slots').split():
src_name = f'_{name}_'
if (value := attrs.pop(src_name, None)) is not None:
spec_attrs[name] = value
if name == 'schema':
spec_attrs['source_schema'] = value
elif name == 'table':
spec_attrs['source_table'] = value
if 'title' not in spec_attrs:
spec_attrs['title'] = clsname
if 'validate' in spec_attrs and spec_attrs['validate'] is True and not VALIDATE_DATA:
raise QuazyError('`validate` attribute is set to `True`, but `pydantic` is not installed')
DB = typing.cast(type[DBTable.DB], type(clsname + 'DB', (cls.db_base_class,), spec_attrs))
DB.self_fields = list()
DB.many_fields = dict()
DB.many_to_many_fields = dict()
DB.defaults = dict()
attrs['DB'] = DB
if DB.use_slots:
attrs['__slots__'] = []
qualname = attrs['__qualname__']
if not DB.table:
DB.source_table = camel2snake(qualname.replace('.', ''))
DB.table = DB.source_table
MetaTable.collect_fields(bases, DB, attrs)
if '.' in qualname:
# save owner class name
chunks = qualname.split('.')
base_cls_name = '.'.join(chunks[:-1])
DB.owner = base_cls_name
field_name = camel2snake(chunks[-1]) + 's'
DB.snake_name = field_name
if field_name in DB.fields:
raise QuazyFieldNameError(f'Subclass name {qualname} repeats (explicit or implicit) field name')
else:
DB.snake_name = camel2snake(qualname) + 's'
DB.subtables = dict()
if 'discriminator' not in spec_attrs:
DB.discriminator = attrs['__qualname__'] if DB.cid else None
new_cls = super().__new__(cls, attrs['__qualname__'], bases, attrs, **kwargs)
"""
if sys.version_info >= (3, 14):
from annotationlib import get_annotate_from_class_namespace, call_annotate_function, Format
annotate_func = get_annotate_from_class_namespace(attrs)
def wrapped_annotate(format):
import annotationlib
annos = call_annotate_function(annotate_func, format, owner=new_cls)
return {key: value for key, value in annos.items() if key in DB.fields}
if annotate_func:
new_cls.__annotate__ = wrapped_annotate
"""
return new_cls
@staticmethod
def collect_fields(bases: tuple[type[DBTable], ...], DB: type[DBTable.DB], attrs: dict[str, Any]):
if DB.extendable:
DB.is_root = True
fields = MetaTable.collect_bases_fields(bases, DB)
has_pk = DB.pk is not None
if sys.version_info < (3, 14):
annotations = attrs.get('__annotations__', {})
else:
annotate_func = annotationlib.get_annotate_from_class_namespace(attrs)
annotations = annotationlib.call_annotate_function(annotate_func, annotationlib.Format.FORWARDREF) if annotate_func else {}
for name, t in annotations.items(): # type: str, type
if name.startswith('_'):
continue
if (isinstance(t, typing._GenericAlias) and t.__name__ == "ClassVar"
or isinstance(t, str) and get_annotated_id(t) == "ClassVar"):
continue
DB.self_fields.append(name)
if DB.use_slots:
attrs['__slots__'].append(name)
if (isinstance(t, typing._AnnotatedAlias) and t.__metadata__[0] == 'ObjVar'
or isinstance(t, str) and get_annotated_id(t) == 'ObjVar'):
if DB.use_slots:
del attrs[name]
attrs['__slots__'].append(name)
continue
if not DB.use_slots:
field = attrs.get(name, DBField())
else:
field = attrs.pop(name, DBField())
if isinstance(field, DBField):
if DB.extendable and not DB.is_root:
field.column = f'{DB.source_table}_{field.column or name}'
if not DB.use_slots:
attrs[name] = field.default
if field.default is not Unassigned:
DB.defaults[name] = field.default
if field.pk:
has_pk = True
DB.pk = field
elif (isinstance(t, typing._AnnotatedAlias) and t.__metadata__[0] == 'FieldCID'
or isinstance(t, str) and get_annotated_id(t) == 'FieldCID'):
# check CID
if not DB.extendable:
raise QuazyFieldTypeError(
f'Table `{attrs["__qualname__"]}` is not declared with `_extendable_` attribute')
elif DB.cid:
raise QuazyFieldTypeError(
f'Table `{attrs["__qualname__"]}` has CID field already inherited from extendable')
field.cid = True
DB.cid = field
elif t is FieldBody or t == 'FieldBody' or field.body:
if DB.body:
raise QuazyFieldTypeError(f'Table `{attrs["__qualname__"]}` has body field already')
field.body = True
DB.body = field
field.prepare(name)
else:
field = DBField(default=field)
field.prepare(name)
field.required = True
DB.defaults[name] = field.default
fields[name] = field
# check seed proper declaration
if DB.cid and not DB.extendable:
raise QuazyFieldTypeError(
f'CID field is declared, but table `{attrs["__qualname__"]}` is not declared with `_extendable_` attribute')
if not has_pk and not DB.meta:
pk = DBField(pk=True)
pk.type = int
pk.prepare('id')
pk.ux.blank = True
fields['id'] = pk
DB.pk = pk
if DB.use_slots:
attrs['__slots__'].append('id')
DB.fields = fields
@staticmethod
def collect_bases_fields(bases: tuple[type, ...], DB: type[DBTable.DB]) -> dict[str, DBField]:
fields: dict[str, DBField] = dict()
for base in bases:
if base is DBTable:
break
if issubclass(base, DBTable):
fields.update(base.DB.fields)
if base.DB.extendable:
if DB.extendable:
raise QuazyNotSupported('Multiple inheritance of extendable tables is not supported')
DB.extendable = base.DB.extendable
DB.is_root = False
DB.cid = base.DB.cid
DB.table = base.DB.table
DB.schema = base.DB.schema
DB.body = base.DB.body
elif base.DB.pk is not None:
DB.pk = base.DB.pk
return fields
def resolve_types(cls, globalns):
"""Resolve fields types from annotations
:meta private:"""
from .db_factory import DBFactory
# eval annotations
if sys.version_info >= (3, 14):
annotations = annotationlib.get_annotations(cls, format=annotationlib.Format.FORWARDREF)
else:
annotations = cls.__annotations__
type_hints = typing.get_type_hints(cls, localns=globalns | {'DBFactory': DBFactory})
for name, t in type_hints.items():
if name not in cls.DB.fields: # or cls.fields[name].type is not None:
continue
field: DBField = cls.DB.fields[name]
annotation = annotations.get(name, None)
if cls.resolve_type(t, annotation, field, globalns):
setattr(cls, name, list())
del cls.DB.fields[name]
# eval owner
if isinstance(cls.DB.owner, str):
base_cls: type[DBTable] = getattr(sys.modules[cls.__module__], cls.DB.owner)
# field_name = camel2snake(cls.__name__)
field = DBField()
field.prepare(base_cls.DB.table)
field.type = base_cls
field.ref = True
field.required = True
cls.DB.owner = base_cls
cls.DB.fields[field.column] = field
# resolve types for subclasses
for name, t in vars(cls).items():
if inspect.isclass(t) and issubclass(t, DBTable):
cls.DB.subtables[t.DB.snake_name] = t
t.DB.schema = cls.DB.schema
t.DB.db = cls.DB.db
t.resolve_types(globalns)
def resolve_type(cls, t: type, ta: type | None, field: DBField, globalns) -> bool | None:
"""Resolve field types from annotations
:meta private:"""
if inspect.isclass(t) and issubclass(t, DBField):
# import custom field attributes
for k, v in t.__dict__.items():
if k.startswith('_'):
continue
if inspect.isclass(v) and issubclass(v, UX):
for kk, vv in v.__dict__.items():
if not kk.startswith('_'):
setattr(field.ux, kk, vv)
else:
setattr(field, k, v)
elif (ta is not None and
(isinstance(ta, typing._AnnotatedAlias) and (meta_name:=ta.__metadata__[0]) or
isinstance(ta, str) and (meta_name:=get_annotated_id(ta)) is not None)):
if meta_name == 'FieldCID':
field.type = t
elif meta_name == 'Property':
field.property = True
field.required = False
cls.resolve_type(t, None, field, globalns)
elif meta_name in ('Many', 'ManyToMany'):
field_type = t.__args__[0]
if not inspect.isclass(field_type) or not issubclass(field_type, DBTable):
raise QuazyFieldTypeError(f'Many type `{t}` should be referenced to another DBTable')
if meta_name == 'Many':
cls.DB.many_fields[field.name] = DBManyField(field_type, field.reverse_name)
else:
cls.DB.many_to_many_fields[field.name] = DBManyToManyField(field_type, field.reverse_name)
return True
elif isinstance(t, typing._UnionGenericAlias) or isinstance(t, types.UnionType):
if len(args:=typing.get_args(t)) == 2 and args[1] is type(None):
# 'Optional' annotation
field.required = False
DBTable.resolve_type(args[0], None, field, globalns)
elif t is FieldBody:
field.type = dict
elif inspect.isclass(t) and issubclass(t, DBTable):
# Foreign key
field.ref = True
field.type = t
if cls.DB.use_slots and field.name not in cls.DB.defaults:
cls.DB.defaults[field.name] = None
elif t in KNOWN_TYPES or inspect.isclass(t) and issubclass(t, Enum):
# Base type
field.type = t
else:
raise QuazyFieldTypeError(f'type `{t}` is not supported as field type for `{cls.__qualname__}.{field.name}`')
def resolve_types_many(cls, add_middle_table: Callable[[type[DBTable]], Any]):
"""Resolve referred types from annotations
:meta private:"""
if cls.DB.meta:
return
# eval refs
for name, field in cls.DB.fields.items(): # type: str, DBField
if field.ref:
rev_name: str = field.reverse_name or cls.DB.snake_name
if rev_name in field.type.DB.many_fields:
if field.type.DB.many_fields[rev_name].foreign_table is not cls:
raise QuazyFieldNameError(
f'Cannot reuse Many field in table `{field.type.__name__}` with name `{rev_name}`, '
f'it is associated with table `{field.type.DB.many_fields[rev_name].foreign_table.__name__}`. '
f'Set different `reverse_name`.')
field.type.DB.many_fields[rev_name].foreign_field = name
else:
field.type.DB.many_fields[rev_name] = DBManyField(cls, name)
# check Many fields connected
for name, m_field in cls.DB.many_fields.items(): # type: str, DBManyField
if not m_field.foreign_field or m_field.foreign_field not in m_field.foreign_table.DB.fields:
raise QuazyFieldTypeError(
f'Cannot find reference from table `{m_field.foreign_table.__name__}` to table `{cls.__name__}` to '
f'connect with Many field `{name}`. Add field to source table or change field type to `ManyToMany`')
# check and connect ManyToMany fields
for name, mm_field in cls.DB.many_to_many_fields.items(): # type: str, DBManyToManyField
if mm_field.middle_table:
continue
middle_table_name = "{}{}".format(cls.__qualname__, name.capitalize())
middle_table_inner_name = "{}_{}".format(cls.DB.table, name)
rev_name = mm_field.foreign_field or cls.DB.snake_name
if rev_name in mm_field.foreign_table.DB.many_to_many_fields and mm_field.foreign_table.DB.many_to_many_fields[
rev_name].foreign_table is not cls:
raise QuazyFieldNameError(
f'Cannot reuse ManyToMany field in table `{mm_field.foreign_table.__name__}` with name `{rev_name}`, '
f'it is associated with table `{mm_field.foreign_table.DB.many_to_many_fields[rev_name].source_table.__name__}`. '
f'Set different `reverse_name`.')
f1 = DBField(mm_field.foreign_table.DB.table, indexed=True)
f1.prepare(f1.column)
f1.type = mm_field.foreign_table
f1.ref = True
f2 = DBField(cls.DB.table, indexed=True)
f2.prepare(f2.column)
f2.type = cls
f2.ref = True
TableClass: type[DBTable] = typing.cast(type[DBTable],
type(middle_table_name, (DBTable,), {
'__qualname__': middle_table_name,
'__module__': cls.__module__,
'__annotate_func__': lambda f: {
f1.name: f1.type,
f2.name: f2.type
},
'__annotations__': {
f1.name: f1.type,
f2.name: f2.type
},
'_table_': middle_table_inner_name,
f1.name: f1,
f2.name: f2,
}))
add_middle_table(TableClass)
mm_field.middle_table = TableClass
mm_field.foreign_field = rev_name
mm_field.foreign_table.DB.many_to_many_fields[rev_name].middle_table = TableClass
mm_field.foreign_table.DB.many_to_many_fields[rev_name].foreign_field = name
[docs]
class DBTable(metaclass=MetaTable):
"""Table model constructor
All class attributes are considered as database table fields.
Types could be set with annotations or/and directly as `DBField` instance.
There are several special class attributes used to set DBTable details and behaviour.
Note:
attribute named `pk` is reserved as property for direct primary key access
"""
# initial attributes
_table_: ClassVar[str] #: database table internal name
_title_: ClassVar[str] #: user-friendly table name
_schema_: ClassVar[str] #: explicit schema name
_just_for_typing_: ClassVar[bool] #: internal flag used for migrations
_extendable_: ClassVar[bool] #: set :ref:`extendable <extendable>` flag for a table
_discriminator_: ClassVar[typing.Any] #: SQL safe CID value to specify table in extended mode
_meta_: ClassVar[bool] #: table marked as :ref:`meta table <meta tables>`
_lookup_field_: ClassVar[str] #: specify field name for text search. For integrations only.
_use_slots_: ClassVar[bool] #: use slots for database fields (enabled by default)
_validate_: ClassVar[bool] #: validate fields types using :mod:`pydantic` (default: True if `pydantic` is installed)
_metadata_: ClassVar[dict[str, typing.Any]] #: any custom metadata
# state attributes
__slots__ = ('_db_', '_modified_fields_')
_db_: DBFactory | None
_modified_fields_: set[str]
[docs]
class DB:
"""DBTable meta-subclass with internal information
It has only class-based attributes, intended for read-only. Instances aren't supported.
"""
db: ClassVar[DBFactory] = None #: :class:`DBFactory` linked to a table, if already specified
table: ClassVar[str | None] = None #: database table name
source_table: ClassVar[str | None] = None #: user defined database table name
title: ClassVar[str | None] = None #: user-friendly table name
schema: ClassVar[str | None] = None #: database schema name
source_schema: ClassVar[str | None] = None #: user-defined schema name
just_for_typing: ClassVar[bool] = False #: internal flag used for migrations
snake_name: ClassVar[str] #: internal flag used for migrations
extendable: ClassVar[bool] = False #: support for :ref:`extendable <extendable>` classes
cid: ClassVar[DBField | None] = None #: CID field reference (if declared)
is_root: ClassVar[bool] = False #: is table a root of extendable tables chain
discriminator: ClassVar[typing.Any | None] = None #: derived table inner code
owner: ClassVar[typing.Union[str, type[DBTable]] | None] = None #: table owner of subtable
subtables: ClassVar[dict[str, type[DBTable]] | None] = None #: subtables list
meta: ClassVar[bool] = False #: table marked as :ref:`meta table <meta tables>`
pk: ClassVar[DBField | None] = None #: reference to primary field :class:`DBField`
body: ClassVar[DBField | None] = None #: reference to :ref:`body field <properties>` or None
many_fields: ClassVar[dict[str, DBManyField] | None] = None #: dict of field sets, when this table is referred from another table
many_to_many_fields: ClassVar[dict[str, DBManyToManyField] | None] = None #: dict of field sets, when two tables referred to each other
fields: ClassVar[dict[str, DBField]] #: all fields dict
self_fields: ClassVar[list[str] | None] #: field names, declared in this class only
defaults: ClassVar[dict[str, typing.Any]] #: default values for fields
lookup_field: ClassVar[str | None] = None #: field name for text search for integrations
use_slots: ClassVar[bool] = False #: use slots for database fields
validate: ClassVar[bool] = VALIDATE_DATA #: validate fields types using :mod:`pydantic`
validators: ClassVar[dict[str, TypeAdapter[Any]]] #: dict of validators for fields
metadata: ClassVar[dict[str, typing.Any]] #: any custom metadata
[docs]
class ItemGetter(typing.Generic[DBTableT]):
def __init__(self, db: DBFactory | DBFactoryAsync, table: type[DBTableT], field_name: str, pk_value: Any, view: str = None):
self._db = db
self._table = table
self._attr_name = field_name
self._pk_value = pk_value
self._view = view
self._cache = dict()
def __str__(self):
return self._view or f'{self._table.__qualname__}[{self._pk_value}]'
def __getattr__(self, item):
if item.startswith('_'):
return super().__getattribute__(self, item)
if item == 'pk' or item == self._table.DB.pk.name:
return self._pk_value
if item in self._cache:
return self._cache[item]
if self._db.async_mode:
raise QuazyWrongOperation("In async mode use `fetch` first")
value = self._db.query(self._table).select(item).filter(pk=self._pk_value).fetch_value()
self._cache[item] = value
return value
def fetch(self, *fields) -> Awaitable[DBTableT] | DBTableT:
actual_fields = fields + ('pk',) if fields else ()
value = self._db.query(self._table).select(*actual_fields).filter(pk=self._pk_value).fetch_one()
self._cache |= value.__dict__
return value
[docs]
class ItemGetterAsync(ItemGetter[DBTableT]):
async def fetch(self, *fields) -> DBTableT:
actual_fields = fields + ('pk',) if fields else ()
value = await self._db.query(self._table).select(*actual_fields).filter(pk=self._pk_value).fetch_one()
self._cache |= value.__dict__
return value
[docs]
class ListGetter(list, typing.Generic[DBTableT]):
def __init__(self, db: DBFactory, table: type[DBTable], field_name: str, pk_value: Any):
self._db = db
self._table = table
self._field_name = field_name
self._pk_value = pk_value
super().__init__()
def fetch(self, *fields) -> Awaitable[list[DBTableT]] | list[DBTableT]:
actual_fields = fields + ('pk',) if fields else ()
self[:] = self._db.query(self._table).select(*actual_fields).filter(lambda x: getattr(x, self._field_name) == self._pk_value).fetch_all()
return self
[docs]
class ListGetterAsync(ListGetter[DBTableT]):
async def fetch(self, *fields) -> list[DBTableT]:
actual_fields = fields + ('pk',) if fields else ()
self[:] = await self._db.query(self._table).select(*actual_fields).filter(lambda x: getattr(x, self._field_name) == self._pk_value).fetch_all()
return self
@classmethod
def setup_validators(cls):
if cls.DB.validate:
validators = {}
for name, field in cls.DB.fields.items():
if field.pk:
adapter = TypeAdapter(field.type | cls, config=PYDANTIC_CONFIG)
elif field.ref:
adapter = TypeAdapter(field.type | field.type.DB.pk.type, config=PYDANTIC_CONFIG)
else:
adapter = TypeAdapter(field.type, config=PYDANTIC_CONFIG)
validators[name] = adapter
cls.DB.validators = validators
for table in cls.DB.subtables.values():
table.setup_validators()
def __pre_init__(self, **initial):
self._modified_fields_: set[str] | None = None
self._db_: DBFactory = initial.pop('_db_', self.DB.db)
self_id = initial.get(self.DB.pk.name, None)
object.__setattr__(self, self.DB.pk.name, None)
if not self.DB.use_slots:
list_getter_cls = DBTable.ListGetter if not self._db_.async_mode else DBTable.ListGetterAsync
for field_name, field in self.DB.many_fields.items():
if self_id is not None:
object.__setattr__(self, field_name, list_getter_cls(self._db_, field.foreign_table, field.foreign_field, self_id))
else:
object.__setattr__(self, field_name, list())
for field_name, subtable in self.DB.subtables.items():
if self_id is not None:
object.__setattr__(self, field_name, list_getter_cls(self._db_, subtable, self.DB.table, self_id))
else:
object.__setattr__(self, field_name, list())
for field_name, field in self.DB.many_to_many_fields.items():
if self_id is not None:
object.__setattr__(self, field_name, list_getter_cls(self._db_, field.foreign_table, field.foreign_field, self_id))
else:
object.__setattr__(self, field_name, list())
def __init__(self, **initial):
"""DBTable instance constructor
Args:
**initial: fields initial values
"""
self.__pre_init__(**initial)
for k, v in initial.items():
if field := self.DB.fields.get(k):
if issubclass(field.type, Enum):
setattr(self, k, field.type(v) if v is not None else None)
continue
# else:
if k not in self.DB.fields and k not in self.DB.many_fields and k not in self.DB.many_to_many_fields:
raise QuazyFieldNameError(f'Wrong field name `{k}` in new instance of `{self.__class__.__name__}`')
setattr(self, k, v)
if self.DB.use_slots:
for k, v in self.DB.defaults.items():
if k not in initial:
if not callable(v):
object.__setattr__(self, k, v)
else:
object.__setattr__(self, k, v(self))
self._modified_fields_ = set(initial.keys())
@classmethod
def raw(cls, **initial) -> Self:
self = cls.__new__(cls)
self.__pre_init__(**initial)
for k, v in initial.copy().items():
if k.endswith("__view"):
continue
if field := self.DB.fields.get(k):
if issubclass(field.type, Enum):
object.__setattr__(self, k, field.type(v) if v is not None else None)
continue
elif field.ref:
view = initial.pop(f'{k}__view', None)
getter_class = DBTable.ItemGetter if not self.DB.db.async_mode else DBTable.ItemGetterAsync
object.__setattr__(self, k, getter_class(self.DB.db, field.type, field.name, v.pk if isinstance(v, DBTable) else v, view))
continue
elif field.property and not cls.DB.db.translator.supports_cast_converter:
v = cls.DB.db.translator.cast_value(field, v)
object.__setattr__(self, k, v)
self._modified_fields_ = set()
return self
def __setattr__(self, key, value):
if key in self.DB.fields:
if self.DB.validate:
try:
from pydantic import ValidationError
except ImportError:
raise
try:
value = self.DB.validators[key].validate_python(value)
except ValidationError as e:
raise QuazyFieldTypeError(f'Field `{key}` in `{self.__class__.__name__}` has wrong type: {e}')
if getattr(self, '_modified_fields_', None) is not None:
self._modified_fields_.add(key)
return super().__setattr__(key, value)
def __class_getitem__(cls, item: Any) -> Awaitable[Self] | Self:
if not cls.DB.db:
raise QuazyWrongOperation(f"Table `{cls.__qualname__}` is not assigned to a database")
return cls.DB.db.get(cls, item)
[docs]
@classmethod
def check_db(cls):
"""Check whether DBTable is assigned to DBFactory
Raises:
QuazyWrongOperation: table is not assigned
:meta private:"""
if not cls.DB.db:
raise QuazyWrongOperation(f"Table `{cls.__qualname__}` is not assigned to a database")
[docs]
@classmethod
def get(cls, pk: Any = None, **fields) -> Awaitable[Self] | Self:
"""Get DBTable instance by primary key value
Args:
pk: primary key value to get an item (optional)
**fields: fields values to find item if no pk is specified (optional)
"""
cls.check_db()
return cls.DB.db.get(cls, pk, **fields)
[docs]
def save(self, **kwargs) -> Awaitable[Self] | Self:
"""Save DBTable instance changes to a database
Args:
kwargs: additional values to update item fields before saving it to the database
"""
self.check_db()
return self._db_.save(self, **kwargs)
[docs]
def load(self, selected_field_name: str | None = None) -> Awaitable[None] | None:
"""Load related items from foreign tables
Args:
selected_field_name: any related field name to load, if not specified, all related fields will be loaded.
"""
self.check_db()
if selected_field_name is not None:
return getattr(self, selected_field_name).fetch()
else:
collection = []
for field_name in itertools.chain(self.DB.subtables, self.DB.many_fields, self.DB.many_to_many_fields):
collection.append(getattr(self, field_name).fetch())
if self._db_.async_mode:
return asyncio.gather(*collection)
return None
[docs]
def delete(self) -> Awaitable[None] | None:
"""Delete DBTable instance from a database"""
self.check_db()
return self._db_.delete(item=self)
[docs]
@classmethod
def query(cls, name: str | None = None) -> DBQuery[Self]:
"""Create a DBQuery instance for queries associated with this table
Args:
name: name of the query for subquery request
Hint:
Use identical method :meth:`select` for your preference.
"""
cls.check_db()
return cls.DB.db.query(cls, name)
[docs]
@classmethod
def select(cls, *field_names: str, **fields: FDBSQL) -> DBQuery[Self]:
"""Create a DBQuery instance and specify selected fields
Read :meth:`DBQuery.select()` for details.
"""
return cls.query().select(*field_names, **fields)
@classmethod
def _dump_schema(cls) -> dict[str, Any]:
res = {
'qualname': cls.__qualname__,
'module': cls.__module__,
'table': cls.DB.table,
'schema': cls.DB.schema,
'fields': {name: f._dump_schema() for name, f in cls.DB.fields.items()},
'self_fields': cls.DB.self_fields if cls.DB.extendable else list(cls.DB.fields),
}
for col in 'extendable discriminator just_for_typing'.split():
if val := getattr(cls.DB, col):
res[col] = val
return res
@classmethod
def _load_schema(cls, state: dict[str, Any]) -> type[DBTable]:
fields = {name: DBField._load_schema(f) for name, f in state['fields'].items()}
self_fields = state['self_fields']
table_class: type[DBTable] = typing.cast(type[DBTable], type(state['qualname'], (DBTable,), {
'__qualname__': state['qualname'],
'__module__': state['module'],
'__annotations__': {name: f.type for name, f in fields.items() if name in self_fields},
'_table_': state['table'],
'_schema_': state['schema'],
'_just_for_typing_': state.get('just_for_typing', False),
'_extendable_': state.get('extendable', False),
'_discriminator_': state.get('discriminator'),
**fields
}))
for name, f in table_class.DB.fields.items():
if f.pk:
table_class.DB.pk = f
elif f.cid:
table_class.DB.cid = f
return table_class
@property
def pk(self):
"""get a primary key value"""
return getattr(self, self.DB.pk.name)
@pk.setter
def pk(self, value):
"""set a primary key value"""
setattr(self, self.DB.pk.name, value)
[docs]
def inspect(self) -> str:
"""Inspect table in simple text format
key: value (type)
"""
res = []
for k, v in vars(self).items():
if not k.startswith('_'):
res.append(f'{k}: {str(v)} ({type(v).__name__})')
return '\n'.join(res)
[docs]
@classmethod
def _view_(cls, item: DBQueryField[typing.Self]):
"""virtual method to override DBTable item presentation
Originally, each table item is requester as a primary key value (integer number for ex.). It is more
convenient to see user-friendly presentation, like `name`, `caption` or several fields combined.
Example::
class User(DBTable):
name: str
@classmethod
def _view_(cls, item: DBQueryField):
return item.name
:meta public:
"""
return None
[docs]
@classmethod
def get_lookup_field(cls, item: DBQueryField) -> DBSQL | None:
"""return lookup field"""
if cls.DB.lookup_field:
return item[cls.DB.lookup_field]
def __eq__(self, other):
return self.pk == other.pk if isinstance(other, DBTable) else other
def __ne__(self, other):
return self.pk != other.pk if isinstance(other, DBTable) else other
def __str__(self):
return f'{self.DB.title}[{self.pk}]'
__repr__ = __str__
[docs]
def _before_update(self, db: DBFactory):
"""abstract event before update to the database"""
[docs]
def _after_update(self, db: DBFactory):
"""abstract event after update to the database"""
[docs]
def _before_insert(self, db: DBFactory):
"""abstract event before insert to the database"""
[docs]
def _after_insert(self, db: DBFactory):
"""abstract event after insert to the database"""
[docs]
def _before_delete(self, db: DBFactory):
"""abstract event before delete from the database"""
[docs]
def _after_delete(self, db: DBFactory):
"""abstract event after delete from the database"""
@typing.overload
def dbtable(
cls=None,
/, *,
table: str = None,
title: str = None,
schema: str = None,
extendable: bool = None,
discriminator: typing.Any = None,
meta: bool = False,
lookup_field: str = None,
use_slots: bool = False,
validate: bool = VALIDATE_DATA,
metadata: dict[str, typing.Any] = None):
...
@typing.overload
def dbtable(cls=None, **kwargs):
...
[docs]
def dbtable(cls=None, **kwargs):
""":func:`~dataclasses.dataclass`-styled decorator for :class:`DBTable` subclasses
Example::
@dbtable
class User:
name: str
"""
def wrapper(cls: type):
attrs = {
f'_{k}_': v
for k, v in kwargs.items()
}
cls_attr = {
k: v
for k, v in vars(cls).items()
if k not in ('__dict__', '__weakref__', '__doc__')
}
return type(
cls.__qualname__,
(DBTable,) + cls.__bases__,
{
'__qualname__': cls.__qualname__,
} | attrs | cls_attr
)
if cls is not None:
return wrapper(cls)
return wrapper