モチベ
Pythonでバックエンドを書いていたら普段Goで書いているのもあってSQL関係で見慣れない記法があったので書く
追記: 意外と奥深くまで潜る必要があったので2回ぐらいしか潜れませんでした。深掘りと言うよりなんとなくあっさり理解するぐらいの記事になってそうです
目標
以下2つのコード内部で何が起きているのかを理解する
import enum
from sqlalchemy import Column, Integer, String, DateTime, func, Enum, Text
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class RoleType(str, enum.Enum):
admin = "admin"
user = "user"
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
username = Column(String(255), nullable=False)
email = Column(String(255), nullable=False)
password = Column(String(255), nullable=False)
role = Column(Enum(RoleType), nullable=False)
created_at = Column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
# User Data取得
users_data = db.query(User).filter(User.id == id).first()
解説
まず
Base = declarative_base()
の部分, これはhttps://github.com/sqlalchemy/sqlalchemy/blob/b39afd5008bef95a8c2c30eada1e22ef6a286670/lib/sqlalchemy/ext/declarative/init.py#L26 に定義が書かれている
@util.moved_20(
"The ``declarative_base()`` function is now available as "
":func:`sqlalchemy.orm.declarative_base`."
)
def declarative_base(*arg, **kw):
return _declarative_base(*arg, **kw)
どうも今はsqlalchemy.orm.decl_apiに移動したらしいが後方互換のために残しているらしい。というわけでそこを見てみると
def declarative_base(
*,
metadata: Optional[MetaData] = None,
mapper: Optional[Callable[..., Mapper[Any]]] = None,
cls: Type[Any] = object,
name: str = "Base",
class_registry: Optional[clsregistry._ClsRegistryType] = None,
type_annotation_map: Optional[_TypeAnnotationMapType] = None,
constructor: Callable[..., None] = _declarative_constructor,
metaclass: Type[Any] = DeclarativeMeta,
) -> Any:
r"""Construct a base class for declarative class definitions.
The new base class will be given a metaclass that produces
appropriate :class:`~sqlalchemy.schema.Table` objects and makes
the appropriate :class:`_orm.Mapper` calls based on the
information provided declaratively in the class and any subclasses
of the class.
.. versionchanged:: 2.0 Note that the :func:`_orm.declarative_base`
function is superseded by the new :class:`_orm.DeclarativeBase` class,
which generates a new "base" class using subclassing, rather than
return value of a function. This allows an approach that is compatible
with :pep:`484` typing tools.
The :func:`_orm.declarative_base` function is a shorthand version
of using the :meth:`_orm.registry.generate_base`
method. That is, the following::
from sqlalchemy.orm import declarative_base
Base = declarative_base()
Is equivalent to::
from sqlalchemy.orm import registry
mapper_registry = registry()
Base = mapper_registry.generate_base()
See the docstring for :class:`_orm.registry`
and :meth:`_orm.registry.generate_base`
for more details.
.. versionchanged:: 1.4 The :func:`_orm.declarative_base`
function is now a specialization of the more generic
:class:`_orm.registry` class. The function also moves to the
``sqlalchemy.orm`` package from the ``declarative.ext`` package.
:param metadata:
An optional :class:`~sqlalchemy.schema.MetaData` instance. All
:class:`~sqlalchemy.schema.Table` objects implicitly declared by
subclasses of the base will share this MetaData. A MetaData instance
will be created if none is provided. The
:class:`~sqlalchemy.schema.MetaData` instance will be available via the
``metadata`` attribute of the generated declarative base class.
:param mapper:
An optional callable, defaults to :class:`_orm.Mapper`. Will
be used to map subclasses to their Tables.
:param cls:
Defaults to :class:`object`. A type to use as the base for the generated
declarative base class. May be a class or tuple of classes.
:param name:
Defaults to ``Base``. The display name for the generated
class. Customizing this is not required, but can improve clarity in
tracebacks and debugging.
:param constructor:
Specify the implementation for the ``__init__`` function on a mapped
class that has no ``__init__`` of its own. Defaults to an
implementation that assigns \**kwargs for declared
fields and relationships to an instance. If ``None`` is supplied,
no __init__ will be provided and construction will fall back to
cls.__init__ by way of the normal Python semantics.
:param class_registry: optional dictionary that will serve as the
registry of class names-> mapped classes when string names
are used to identify classes inside of :func:`_orm.relationship`
and others. Allows two or more declarative base classes
to share the same registry of class names for simplified
inter-base relationships.
:param type_annotation_map: optional dictionary of Python types to
SQLAlchemy :class:`_types.TypeEngine` classes or instances. This
is used exclusively by the :class:`_orm.MappedColumn` construct
to produce column types based on annotations within the
:class:`_orm.Mapped` type.
.. versionadded:: 2.0
.. seealso::
:ref:`orm_declarative_mapped_column_type_map`
:param metaclass:
Defaults to :class:`.DeclarativeMeta`. A metaclass or __metaclass__
compatible callable to use as the meta type of the generated
declarative base class.
.. seealso::
:class:`_orm.registry`
"""
return registry(
metadata=metadata,
class_registry=class_registry,
constructor=constructor,
type_annotation_map=type_annotation_map,
).generate_base(
mapper=mapper,
cls=cls,
name=name,
metaclass=metaclass,
)
どうもこれはSQLAlchemyで使われる「宣言型クラス」を定義するための基盤クラスを構築する関数で, この基盤クラスを利用して、Pythonクラスとデータベーステーブルのマッピングを記述できるらしい。バージョンアップもあり
from sqlalchemy.orm import declarative_base
Base = declarative_base()
はより汎用的なクラスとしてregistryクラスが導入され、
from sqlalchemy.orm import registry
mapper_registry = registry()
Base = mapper_registry.generate_base()
の簡略版になったらしい。最終的に, デフォルトのままだとdeclative_baseはDeclarativeMetaクラスを呼び出している。
def generate_base(
self,
mapper: Optional[Callable[..., Mapper[Any]]] = None,
cls: Type[Any] = object,
name: str = "Base",
metaclass: Type[Any] = DeclarativeMeta,
) -> Any:
## 省略
return metaclass(name, bases, class_dict)
class DeclarativeMeta(DeclarativeAttributeIntercept):
metadata: MetaData
registry: RegistryType
def __init__(
cls, classname: Any, bases: Any, dict_: Any, **kw: Any
) -> None:
# use cls.__dict__, which can be modified by an
# __init_subclass__() method (#7900)
dict_ = cls.__dict__
# early-consume registry from the initial declarative base,
# assign privately to not conflict with subclass attributes named
# "registry"
reg = getattr(cls, "_sa_registry", None)
if reg is None:
reg = dict_.get("registry", None)
if not isinstance(reg, registry):
raise exc.InvalidRequestError(
"Declarative base class has no 'registry' attribute, "
"or registry is not a sqlalchemy.orm.registry() object"
)
else:
cls._sa_registry = reg
if not cls.__dict__.get("__abstract__", False):
_as_declarative(reg, cls, dict_)
type.__init__(cls, classname, bases, dict_)
次に, User class部分の解説を行う。User classはデフォルトだとBase=DeclarativeMetaクラスを継承している
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
username = Column(String(255), nullable=False)
email = Column(String(255), nullable=False)
password = Column(String(255), nullable=False)
role = Column(Enum(RoleType), nullable=False)
created_at = Column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
継承の際, DeclarativeMetaクラスの中身をdebuggerで見てみると
cls.__dict__ = {'__module__': 'app.models.sqlalchemy.models',
'__tablename__': 'users',
'id': Column(None, Integer(), table=None, primary_key=True, nullable=False),
'username': Column(None, String(length=255), table=None, nullable=False),
'email': Column(None, String(length=255), table=None, nullable=False),
'password': Column(None, String(length=255), table=None, nullable=False),
'role': Column(None, Enum('admin', 'user', name='roletype'), table=None, nullable=False),
'created_at': Column(None, DateTime(timezone=True), table=None, nullable=False, server_default=DefaultClause(<sqlalchemy.sql.functions.now at 0x1089bc4a0; now>, for_update=False)),
'__doc__': None}
であることがわかった。このあと_MapperConfig classに渡されDatabaseとのマッピング処理をしていた。
次にdb部分の処理の解説を行う。dbは実際にはSessionクラスであり, SQLAlchemyのソースコードではここにある
# User Data取得
users_data = db.query(User).filter(User.id == id).first()
まずSessionクラスはqueryメソッドを持っており, QueryクラスかRowReturningQueryを返していた。
@overload
def query(self, _entity: _EntityType[_O]) -> Query[_O]: ...
@overload
def query(
self, _colexpr: TypedColumnsClauseRole[_T]
) -> RowReturningQuery[_T]: ...
# START OVERLOADED FUNCTIONS self.query RowReturningQuery 2-8
# code within this block is **programmatically,
# statically generated** by tools/generate_tuple_map_overloads.py
@overload
def query(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], /
) -> RowReturningQuery[_T0, _T1]: ...
RowReturningQueryクラスはQueryクラスの継承先なので, 結局Queryクラスを見れば良い
class RowReturningQuery(Query[Row[Unpack[_Ts]]]):
if TYPE_CHECKING:
def tuples(self) -> Query[Tuple[Unpack[_Ts]]]: # type: ignore
...
class Query(
_SelectFromElements,
SupportsCloneAnnotations,
HasPrefixes,
HasSuffixes,
HasHints,
EventTarget,
log.Identified,
Generative,
Executable,
Generic[_T],
):
"""ORM-level SQL construction object.
.. legacy:: The ORM :class:`.Query` object is a legacy construct
as of SQLAlchemy 2.0. See the notes at the top of
:ref:`query_api_toplevel` for an overview, including links to migration
documentation.
:class:`_query.Query` objects are normally initially generated using the
:meth:`~.Session.query` method of :class:`.Session`, and in
less common cases by instantiating the :class:`_query.Query` directly and
associating with a :class:`.Session` using the
:meth:`_query.Query.with_session`
method.
Qeury classの中にfilter, all, firstなどのメソッドが全てあり, filterメソッドで絞り込みを行い, そのあとallとfirstで元のUser classから条件に当てはまるものを返していた。
def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
# 中略
for crit in list(criterion):
crit = coercions.expect(
roles.WhereHavingRole, crit, apply_propagate_attrs=self
)
self._where_criteria += (crit,)
return self
def all(self) -> List[_T]:
"""Return the results represented by this :class:`_query.Query`
as a list.
This results in an execution of the underlying SQL statement.
.. warning:: The :class:`_query.Query` object,
when asked to return either
a sequence or iterator that consists of full ORM-mapped entities,
will **deduplicate entries based on primary key**. See the FAQ for
more details.
.. seealso::
:ref:`faq_query_deduplicating`
.. seealso::
:meth:`_engine.Result.all` - v2 comparable method.
:meth:`_engine.Result.scalars` - v2 comparable method.
"""
return self._iter().all() # type: ignore
ところで, filter methodは実際に使う際に
users_data = db.query(User).filter(User.id == id).first()
として呼び出されるのだが, これがどうもfilterの引数には==演算子の結果が入っているっぽいのが気になる。ここにデバッガを適用すると, User.id == idのタイプは
<sqlalchemy.sql.elements.BinaryExpression object at 0x1049c8890>
らしいことがわかる。定義はここにあった
中身をざっくりみると, left, right, operatorのpropertyがそれぞれあり, なんらかの処理を挟んでそうであった。
ここで満足したのでやめる