モチベ

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がそれぞれあり, なんらかの処理を挟んでそうであった。

ここで満足したのでやめる