コンテンツにスキップ

pytilpack.sqlalchemy

必要なextra

pip install pytilpack[sqlalchemy]

pytilpack.sqlalchemy

SQLAlchemy用のユーティリティ集。

AsyncMixin

Bases: AsyncAttrs, _ReprMixin, _ToDictMixin

モデルのベースクラス。SQLAlchemy 2.0スタイル・async前提。

例：

モデル定義例::

class Base(sqlalchemy.orm.DeclarativeBase, pytilpack.sqlalchemy_.AsyncMixin):
    pass

class User(Base):
    __tablename__ = "users"
    ...

Quart例::

@app.before_request
async def _before_request() -> None:
    quart.g.db_session_token = await models.Base.start_session()

@app.teardown_request
async def _teardown_request(_: BaseException | None) -> None:
    if hasattr(quart.g, "db_session_token"):
        await models.Base.close_session(quart.g.db_session_token)
        del quart.g.db_session_token

session_var = contextvars.ContextVar('session_var') class-attribute instance-attribute

セッション。

init(url, pool_size=None, max_overflow=None, pool_recycle=280, pool_pre_ping=True, autoflush=True, expire_on_commit=False, eagerly_init=True, **kwargs) classmethod

DB接続を初期化する。

engineとsessionmakerはスレッドローカルで遅延初期化される。 各スレッドが初めてDB操作を行う際にそのスレッド専用のengineとsessionmakerが生成される。

引数：

名前 タイプ デスクリプション デフォルト
url str | URL

DB接続URL。

 必須
pool_size int | None

コネクションプールのサイズ。スレッド数に応じて調整要。負数の場合はプーリングを無効化する。

 None
max_overflow int | None

コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。

 None
pool_recycle int | None

コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。

 280
pool_pre_ping bool

コネクションプールのプレピン。Noneの場合はデフォルト値を使用。

 True
autoflush bool

セッションのautoflushフラグ。デフォルトはTrue。

 True
expire_on_commit bool

セッションのexpire_on_commitフラグ。デフォルトはFalse。

 False
eagerly_init bool

Trueの場合（デフォルト）、呼び出し元スレッドのengineとsessionmakerをすぐに生成する。

 True
**kwargs Any

その他のsqlalchemy.create_async_engine()へのキーワード引数。

 {}
ソースコード位置： pytilpack/sqlalchemy/async_.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@classmethod
def init(
    cls,
    url: str | sqlalchemy.engine.URL,
    pool_size: int | None = None,
    max_overflow: int | None = None,
    pool_recycle: int | None = 280,
    pool_pre_ping: bool = True,
    autoflush: bool = True,
    expire_on_commit: bool = False,
    eagerly_init: bool = True,
    **kwargs: typing.Any,
) -> None:
    """DB接続を初期化する。

    engineとsessionmakerはスレッドローカルで遅延初期化される。
    各スレッドが初めてDB操作を行う際にそのスレッド専用のengineとsessionmakerが生成される。

    Args:
        url: DB接続URL。
        pool_size: コネクションプールのサイズ。スレッド数に応じて調整要。負数の場合はプーリングを無効化する。
        max_overflow: コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。
        pool_recycle: コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。
        pool_pre_ping: コネクションプールのプレピン。Noneの場合はデフォルト値を使用。
        autoflush: セッションのautoflushフラグ。デフォルトはTrue。
        expire_on_commit: セッションのexpire_on_commitフラグ。デフォルトはFalse。
        eagerly_init: Trueの場合（デフォルト）、呼び出し元スレッドのengineとsessionmakerをすぐに生成する。
        **kwargs: その他のsqlalchemy.create_async_engine()へのキーワード引数。

    """
    if cls._init_args is not None:
        raise RuntimeError("DB接続はすでに初期化されています。")

    if pool_size is not None and pool_size < 0:
        engine_kwargs = kwargs.copy()
        engine_kwargs["poolclass"] = sqlalchemy.pool.NullPool
    else:
        if pool_size is not None and max_overflow is None:
            max_overflow = pool_size * 1  # デフォルトで倍まで許可
        engine_kwargs = kwargs.copy()
        if pool_size is not None:
            engine_kwargs["pool_size"] = pool_size
        if max_overflow is not None:
            engine_kwargs["max_overflow"] = max_overflow
    if pool_recycle is not None:
        engine_kwargs["pool_recycle"] = pool_recycle
    if pool_pre_ping is not None:
        engine_kwargs["pool_pre_ping"] = pool_pre_ping

    cls._init_args = _InitArgs(
        url=url,
        autoflush=autoflush,
        expire_on_commit=expire_on_commit,
        engine_kwargs=engine_kwargs,
    )
    cls._thread_local = threading.local()

    if eagerly_init:
        cls._get_engine_and_sessionmaker()

term() async classmethod

現在のスレッドのDB接続を終了する。

スレッドローカルなengineをdisposeして解放する。 スレッド終了時やテスト後の後始末に使用する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@classmethod
async def term(cls) -> None:
    """現在のスレッドのDB接続を終了する。

    スレッドローカルなengineをdisposeして解放する。
    スレッド終了時やテスト後の後始末に使用する。

    """
    if cls._init_args is None:
        logger.warning("未初期化時のterm()呼び出し")
        return
    if cls._thread_local is None:
        return
    if hasattr(cls._thread_local, "engine") and cls._thread_local.engine is not None:
        await cls._thread_local.engine.dispose()
        cls._thread_local.engine = None
        cls._thread_local.sessionmaker = None

engine() classmethod

現在のスレッドのDB接続を返す。init()後に初めて呼ばれた時点で遅延生成される。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
170
171
172
173
@classmethod
def engine(cls) -> sqlalchemy.ext.asyncio.AsyncEngine:
    """現在のスレッドのDB接続を返す。init()後に初めて呼ばれた時点で遅延生成される。"""
    return cls._get_engine_and_sessionmaker()[0]

sessionmaker() classmethod

現在のスレッドのセッションファクトリを返す。init()後に初めて呼ばれた時点で遅延生成される。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
175
176
177
178
179
180
@classmethod
def sessionmaker(
    cls,
) -> sqlalchemy.ext.asyncio.async_sessionmaker[sqlalchemy.ext.asyncio.AsyncSession]:
    """現在のスレッドのセッションファクトリを返す。init()後に初めて呼ばれた時点で遅延生成される。"""
    return cls._get_engine_and_sessionmaker()[1]

connect() classmethod

DBに接続する。

使用例:: async with Base.connect() as conn: await conn.run_sync(Base.metadata.create_all)

ソースコード位置： pytilpack/sqlalchemy/async_.py 
182
183
184
185
186
187
188
189
190
191
@classmethod
def connect(cls) -> sqlalchemy.ext.asyncio.AsyncConnection:
    """DBに接続する。

    使用例::
        async with Base.connect() as conn:
            await conn.run_sync(Base.metadata.create_all)

    """
    return cls.engine().connect()

session_scope(name=None, log_level=logging.DEBUG) async classmethod

セッションを開始するコンテキストマネージャ。

使用例:: async with Base.session_scope() as session: ...

引数：

名前 タイプ デスクリプション デフォルト
name str | None

セッション名。指定時のみログ出力する。

 None
log_level int

ログレベル。

 DEBUG
ソースコード位置： pytilpack/sqlalchemy/async_.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@classmethod
@contextlib.asynccontextmanager
async def session_scope(
    cls,
    name: str | None = None,
    log_level: int = logging.DEBUG,
):
    """セッションを開始するコンテキストマネージャ。

    使用例::
        async with Base.session_scope() as session:
            ...

    Args:
        name: セッション名。指定時のみログ出力する。
        log_level: ログレベル。

    """
    token = await cls.start_session(name=name, log_level=log_level)
    try:
        yield cls.session()
    finally:
        await cls.close_session(token, name=name, log_level=log_level)

start_session(name=None, log_level=logging.DEBUG) async classmethod

セッションを開始する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@classmethod
async def start_session(
    cls, name: str | None = None, log_level: int = logging.DEBUG
) -> contextvars.Token[sqlalchemy.ext.asyncio.AsyncSession]:
    """セッションを開始する。"""
    session = cls.sessionmaker()()  # pylint: disable=not-callable
    token = cls.session_var.set(session)
    if name is not None:
        logger.log(
            log_level,
            f"セッション開始: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    return token

close_session(token, name=None, log_level=logging.DEBUG) async classmethod

セッションを終了する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@classmethod
async def close_session(
    cls,
    token: contextvars.Token[sqlalchemy.ext.asyncio.AsyncSession],
    name: str | None = None,
    log_level: int = logging.DEBUG,
) -> None:
    """セッションを終了する。"""
    session = cls.session()
    if name is not None:
        logger.log(
            log_level,
            f"セッション終了: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    await asafe_close(session)
    cls.session_var.reset(token)

session() classmethod

セッションを取得する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
252
253
254
255
256
257
258
@classmethod
def session(cls) -> sqlalchemy.ext.asyncio.AsyncSession:
    """セッションを取得する。"""
    sess = cls.session_var.get(None)
    if sess is None:
        raise RuntimeError(f"セッションが開始されていません。{cls.__qualname__}.start_session()を呼び出してください。")
    return sess

select() classmethod

sqlalchemy.Selectを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
260
261
262
263
264
@classmethod
def select(cls) -> sqlalchemy.Select[tuple[typing.Self]]:
    """sqlalchemy.Selectを返す。"""
    # cls.count()などでfrom句が消えないように明示的にfrom句を指定して返す。
    return sqlalchemy.select(cls).select_from(cls)

insert() classmethod

sqlalchemy.Insertを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
266
267
268
269
@classmethod
def insert(cls) -> sqlalchemy.Insert:
    """sqlalchemy.Insertを返す。"""
    return sqlalchemy.insert(cls)

update() classmethod

sqlalchemy.Updateを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
271
272
273
274
@classmethod
def update(cls) -> sqlalchemy.Update:
    """sqlalchemy.Updateを返す。"""
    return sqlalchemy.update(cls)

delete() classmethod

sqlalchemy.Deleteを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
276
277
278
279
@classmethod
def delete(cls) -> sqlalchemy.Delete:
    """sqlalchemy.Deleteを返す。"""
    return sqlalchemy.delete(cls)

count(query) async classmethod

queryのレコード数を取得する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
281
282
283
284
285
286
287
288
289
290
@classmethod
async def count(cls, query: sqlalchemy.Select | sqlalchemy.CompoundSelect) -> int:
    """queryのレコード数を取得する。"""
    # pylint: disable=not-callable
    return (
        await cls.scalar_one_or_none(
            sqlalchemy.select(sqlalchemy.func.count()).select_from(query.order_by(None).subquery())
        )
        or 0
    )

scalar_one(query) async classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@classmethod
async def scalar_one[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).scalar_one()

scalar_one_or_none(query) async classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
@classmethod
async def scalar_one_or_none[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).scalar_one_or_none()

scalars(query) async classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[T]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
325
326
327
328
329
330
331
332
333
334
335
336
@classmethod
async def scalars[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> list[T]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list((await cls.session().execute(query)).scalars().all())

one(query) async classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT]

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
@classmethod
async def one[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT]:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).one()

one_or_none(query) async classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT] | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
@classmethod
async def one_or_none[TT: tuple](
    cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]
) -> sqlalchemy.Row[TT] | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).one_or_none()

all(query) async classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[Row[TT]]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
373
374
375
376
377
378
379
380
381
382
383
384
@classmethod
async def all[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> list[sqlalchemy.Row[TT]]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list((await cls.session().execute(query)).all())

get_by_id_not_null(id_, for_update=False, options=None) async classmethod

IDを元にインスタンスを取得。見つからない場合は例外を送出する。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self

インスタンス。

Raises: ValueError: 見つからない場合。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
@classmethod
async def get_by_id_not_null(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self:
    """IDを元にインスタンスを取得。見つからない場合は例外を送出する。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。
    Raises:
        ValueError: 見つからない場合。
    """
    instance = await cls.get_by_id(id_, for_update=for_update, options=options)
    if instance is None:
        raise ValueError(f"{cls.__qualname__}が見つかりませんでした。id={id_}")
    return instance

get_by_id(id_, for_update=False, options=None) async classmethod

IDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
@classmethod
async def get_by_id(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self | None:
    """IDを元にインスタンスを取得。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。

    """
    q = cls.select().where(cls.id == id_)  # type: ignore  # pylint: disable=no-member
    if options is not None:
        q = q.options(options)
    if for_update:
        q = q.with_for_update()
    return await cls.scalar_one_or_none(q)  # type: ignore[arg-type]

paginate(query, page, per_page, scalar=True) async classmethod

Flask-SQLAlchemy風ページネーション。

引数：

名前 タイプ デスクリプション デフォルト
query Select | CompoundSelect

ページネーションするクエリ。

 必須
page int

ページ番号。

 必須
per_page int

1ページあたりのアイテム数。

 必須
scalar bool

Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

 True

戻り値：

タイプ デスクリプション
Paginator

ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
@classmethod
async def paginate(
    cls,
    query: sqlalchemy.Select | sqlalchemy.CompoundSelect,
    page: int,
    per_page: int,
    scalar: bool = True,
) -> pytilpack.paginator.Paginator:
    """Flask-SQLAlchemy風ページネーション。

    Args:
        query: ページネーションするクエリ。
        page: ページ番号。
        per_page: 1ページあたりのアイテム数。
        scalar: Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

    Returns:
        ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
    """
    if page <= 0:
        raise ValueError(f"ページ番号は1以上でなければなりません: page={page}")
    if per_page <= 0:
        raise ValueError(f"1ページあたりのアイテム数は1以上でなければなりません: per_page={per_page}")
    total = await cls.count(query)
    page_query = query.offset((page - 1) * per_page).limit(per_page)
    items = await (cls.scalars(page_query) if scalar else cls.all(page_query))
    # pylint: disable=protected-access
    return pytilpack.paginator.Paginator(page=page, per_page=per_page, items=items, total=total)

run_with_session(func, *args, **kwargs) classmethod

非同期関数をセッション付きで同期実行する関数。

引数：

名前 タイプ デスクリプション デフォルト
func Callable[P, Awaitable[R]]

デコレート対象の非同期関数

 必須
*args P.args

非同期関数への引数

 ()
**kwargs P.kwargs

非同期関数へのキーワード引数

 {}

戻り値：

タイプ デスクリプション
R

非同期関数の戻り値
ソースコード位置： pytilpack/sqlalchemy/async_.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
@classmethod
def run_with_session[**P, R](cls, func: typing.Callable[P, typing.Awaitable[R]], *args: P.args, **kwargs: P.kwargs) -> R:
    """非同期関数をセッション付きで同期実行する関数。

    Args:
        func: デコレート対象の非同期関数
        *args: 非同期関数への引数
        **kwargs: 非同期関数へのキーワード引数

    Returns:
        非同期関数の戻り値
    """

    async def wrapper() -> R:
        async with cls.session_scope():
            return await func(*args, **kwargs)

    return pytilpack.asyncio.run(wrapper())

AsyncUniqueIDMixin

self.unique_idを持つテーブルクラスにユニークID操作メソッドを付与するmixin。

generate_unique_id() classmethod

ユニークIDを生成する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
481
482
483
484
@classmethod
def generate_unique_id(cls) -> str:
    """ユニークIDを生成する。"""
    return secrets.token_urlsafe(32)

get_by_unique_id(unique_id, allow_id=False, for_update=False) async classmethod

ユニークIDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
unique_id str | int

ユニークID。

 必須
allow_id bool

ユニークIDだけでなくID(int)も許可するかどうか。

 False
for_update bool

更新ロックを取得するか否か。

 False

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
@classmethod
async def get_by_unique_id(
    cls: type[typing.Self],
    unique_id: str | int,
    allow_id: bool = False,
    for_update: bool = False,
) -> typing.Self | None:
    """ユニークIDを元にインスタンスを取得。

    Args:
        unique_id: ユニークID。
        allow_id: ユニークIDだけでなくID(int)も許可するかどうか。
        for_update: 更新ロックを取得するか否か。

    Returns:
        インスタンス。

    """
    assert issubclass(cls, AsyncMixin)
    if allow_id and isinstance(unique_id, int):
        q = cls.select().where(cls.id == unique_id)  # type: ignore
    else:
        q = cls.select().where(cls.unique_id == unique_id)  # type: ignore
    if for_update:
        q = q.with_for_update()
    return await cls.scalar_one_or_none(q)

Mixin

Bases: _ReprMixin, _ToDictMixin

テーブルクラスに共通メソッドを付与するMixin。

get_by_id_not_null(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。見つからない場合は例外を送出する。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self

インスタンス。

Raises: ValueError: 見つからない場合。

ソースコード位置： pytilpack/sqlalchemy/flask.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
def get_by_id_not_null(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self:
    """IDを元にインスタンスを取得。見つからない場合は例外を送出する。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。
    Raises:
        ValueError: 見つからない場合。
    """
    instance = cls.get_by_id(id_, for_update=for_update, options=options)
    if instance is None:
        raise ValueError(f"{cls.__qualname__}が見つかりませんでした。id={id_}")
    return instance

get_by_id(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用。

 None

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/flask.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@classmethod
def get_by_id(
    cls: type[typing.Self], id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self | None:
    """IDを元にインスタンスを取得。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用。

    Returns:
        インスタンス。

    """
    q = cls.query.filter(cls.id == id_)  # type: ignore
    if options is not None:
        q = q.options(options)
    if for_update:
        q = q.with_for_update()
    return q.one_or_none()

UniqueIDMixin

self.unique_idを持つテーブルクラスにユニークID操作メソッドを付与するmixin。

generate_unique_id() classmethod

ユニークIDを生成する。

ソースコード位置： pytilpack/sqlalchemy/flask.py 
84
85
86
87
@classmethod
def generate_unique_id(cls) -> str:
    """ユニークIDを生成する。"""
    return secrets.token_urlsafe(32)

get_by_unique_id(unique_id, allow_id=False, for_update=False) classmethod

ユニークIDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
unique_id str | int

ユニークID。

 必須
allow_id bool

ユニークIDだけでなくID(int)も許可するかどうか。

 False
for_update bool

更新ロックを取得するか否か。

 False

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/flask.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_by_unique_id(
    cls: type[typing.Self],
    unique_id: str | int,
    allow_id: bool = False,
    for_update: bool = False,
) -> typing.Self | None:
    """ユニークIDを元にインスタンスを取得。

    Args:
        unique_id: ユニークID。
        allow_id: ユニークIDだけでなくID(int)も許可するかどうか。
        for_update: 更新ロックを取得するか否か。

    Returns:
        インスタンス。

    """
    if allow_id and isinstance(unique_id, int):
        q = cls.query.filter(cls.id == unique_id)  # type: ignore
    else:
        q = cls.query.filter(cls.unique_id == unique_id)  # type: ignore
    if for_update:
        q = q.with_for_update()
    return q.one_or_none()

SyncAsyncBridgeMixin

同期DBを非同期環境から扱うためのブリッジMixin。

SQLAlchemyの同期APIによるI/Oをasyncio.to_threadでスレッドに逃がし、 awaitableなインターフェースを提供する。各呼び出しは別スレッド・独立セッション ・独立トランザクションで実行される。

SyncMixinが継承する形で提供されるため、SyncMixin.acount(...)等の 呼び出しはそのまま維持される。非同期専用のコードから同期DBを利用する場合の 責務分離を明確にする目的で、別Mixinとして切り出している。

acount(query) async classmethod

queryのレコード数を返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
38
39
40
41
42
43
@classmethod
async def acount(cls, query: sqlalchemy.Select | sqlalchemy.CompoundSelect) -> int:
    """queryのレコード数を返す。非同期版。"""
    # cls.count.__func__はSyncMixin側で定義されるためmypy向けに抑制する。
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.count.__func__)(sync_cls, query)  # type: ignore[attr-defined]

ascalar_one(query) async classmethod

queryの結果を1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
45
46
47
48
49
@classmethod
async def ascalar_one[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T:
    """queryの結果を1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.scalar_one.__func__)(sync_cls, query)  # type: ignore[attr-defined]

ascalar_one_or_none(query) async classmethod

queryの結果を0件または1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
51
52
53
54
55
@classmethod
async def ascalar_one_or_none[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T | None:
    """queryの結果を0件または1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.scalar_one_or_none.__func__)(sync_cls, query)  # type: ignore[attr-defined]

ascalars(query) async classmethod

queryの結果を全件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
57
58
59
60
61
@classmethod
async def ascalars[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> list[T]:
    """queryの結果を全件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.scalars.__func__)(sync_cls, query)  # type: ignore[attr-defined]

aone(query) async classmethod

queryの結果を1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
63
64
65
66
67
@classmethod
async def aone[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT]:
    """queryの結果を1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.one.__func__)(sync_cls, query)  # type: ignore[attr-defined]

aone_or_none(query) async classmethod

queryの結果を0件または1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
69
70
71
72
73
74
75
@classmethod
async def aone_or_none[TT: tuple](
    cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]
) -> sqlalchemy.Row[TT] | None:
    """queryの結果を0件または1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.one_or_none.__func__)(sync_cls, query)  # type: ignore[attr-defined]

aall(query) async classmethod

queryの結果を全件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
77
78
79
80
81
@classmethod
async def aall[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> list[sqlalchemy.Row[TT]]:
    """queryの結果を全件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.all.__func__)(sync_cls, query)  # type: ignore[attr-defined]

aget_by_id(id_, for_update=False) async classmethod

IDを元にインスタンスを取得する。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
83
84
85
86
87
@classmethod
async def aget_by_id(cls, id_: int, for_update: bool = False) -> typing.Self | None:
    """IDを元にインスタンスを取得する。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.get_by_id.__func__)(sync_cls, id_, for_update)  # type: ignore[attr-defined,return-value]

apaginate(query, page, per_page, scalar=True) async classmethod

Flask-SQLAlchemy風ページネーションを返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
89
90
91
92
93
94
95
96
97
98
99
@classmethod
async def apaginate(
    cls,
    query: sqlalchemy.Select | sqlalchemy.CompoundSelect,
    page: int,
    per_page: int,
    scalar: bool = True,
) -> pytilpack.paginator.Paginator:
    """Flask-SQLAlchemy風ページネーションを返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.paginate.__func__)(sync_cls, query, page, per_page, scalar)  # type: ignore[attr-defined]

SyncMixin

Bases: _ReprMixin, _ToDictMixin, SyncAsyncBridgeMixin

モデルのベースクラス。SQLAlchemy 2.0スタイル・同期前提。

例：

モデル定義例::

class Base(sqlalchemy.orm.DeclarativeBase, pytilpack.sqlalchemy_.sync_.SyncMixin):
    pass

class User(Base):
    __tablename__ = "users"
    ...

Quart例::

@app.before_request
async def _before_request() -> None:
    quart.g.db_session_token = models.Base.start_session()

@app.teardown_request
async def _teardown_request(_: BaseException | None) -> None:
    if hasattr(quart.g, "db_session_token"):
        models.Base.close_session(quart.g.db_session_token)
        del quart.g.db_session_token

engine = None class-attribute instance-attribute

DB接続。

sessionmaker = None class-attribute instance-attribute

セッションファクトリ。

session_var = contextvars.ContextVar('session_var') class-attribute instance-attribute

セッション。

init(url, pool_size=None, max_overflow=None, pool_recycle=280, pool_pre_ping=True, autoflush=True, expire_on_commit=False, **kwargs) classmethod

DB接続を初期化する。

引数：

名前 タイプ デスクリプション デフォルト
url str | URL

DB接続URL。

 必須
pool_size int | None

コネクションプールのサイズ。スレッド数に応じて調整要。

 None
max_overflow int | None

コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。

 None
pool_recycle int | None

コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。

 280
pool_pre_ping bool

コネクションプールのプレピン。Noneの場合はデフォルト値を使用。

 True
autoflush bool

セッションのautoflushフラグ。デフォルトはTrue。

 True
expire_on_commit bool

セッションのexpire_on_commitフラグ。デフォルトはFalse。

 False
**kwargs Any

sqlalchemy.create_engineに渡す追加のキーワード引数。

 {}
ソースコード位置： pytilpack/sqlalchemy/sync.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@classmethod
def init(
    cls,
    url: str | sqlalchemy.engine.URL,
    pool_size: int | None = None,
    max_overflow: int | None = None,
    pool_recycle: int | None = 280,
    pool_pre_ping: bool = True,
    autoflush: bool = True,
    expire_on_commit: bool = False,
    **kwargs: typing.Any,
) -> None:
    """DB接続を初期化する。

    Args:
        url: DB接続URL。
        pool_size: コネクションプールのサイズ。スレッド数に応じて調整要。
        max_overflow: コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。
        pool_recycle: コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。
        pool_pre_ping: コネクションプールのプレピン。Noneの場合はデフォルト値を使用。
        autoflush: セッションのautoflushフラグ。デフォルトはTrue。
        expire_on_commit: セッションのexpire_on_commitフラグ。デフォルトはFalse。
        **kwargs: sqlalchemy.create_engineに渡す追加のキーワード引数。

    """
    if cls.engine is not None:
        raise RuntimeError("DB接続はすでに初期化されています。")

    if pool_size is not None and max_overflow is None:
        max_overflow = pool_size * 1  # デフォルトで倍まで許可
    kwargs = kwargs.copy()
    if pool_size is not None:
        kwargs["pool_size"] = pool_size
    if max_overflow is not None:
        kwargs["max_overflow"] = max_overflow
    if pool_recycle is not None:
        kwargs["pool_recycle"] = pool_recycle
    if pool_pre_ping is not None:
        kwargs["pool_pre_ping"] = pool_pre_ping

    cls.engine = sqlalchemy.create_engine(url, **kwargs)
    atexit.register(cls.engine.dispose)

    cls.sessionmaker = sqlalchemy.orm.sessionmaker(cls.engine, autoflush=autoflush, expire_on_commit=expire_on_commit)

connect() classmethod

DBに接続する。

使用例:: with Base.connect() as conn: Base.metadata.create_all(conn)

ソースコード位置： pytilpack/sqlalchemy/sync.py 
183
184
185
186
187
188
189
190
191
192
193
194
@classmethod
def connect(cls) -> sqlalchemy.Connection:
    """DBに接続する。

    使用例::
        with Base.connect() as conn:
            Base.metadata.create_all(conn)

    """
    if cls.engine is None:
        raise RuntimeError("init()が呼ばれていません。")
    return cls.engine.connect()

session_scope(name=None, log_level=logging.DEBUG) classmethod

セッションを開始するコンテキストマネージャ。

使用例:: with Base.session_scope() as session: ...

引数：

名前 タイプ デスクリプション デフォルト
name str | None

セッション名。指定時のみログ出力する。

 None
log_level int

ログレベル。

 DEBUG
ソースコード位置： pytilpack/sqlalchemy/sync.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@classmethod
@contextlib.contextmanager
def session_scope(
    cls,
    name: str | None = None,
    log_level: int = logging.DEBUG,
) -> typing.Generator[sqlalchemy.orm.Session, None, None]:
    """セッションを開始するコンテキストマネージャ。

    使用例::
        with Base.session_scope() as session:
            ...

    Args:
        name: セッション名。指定時のみログ出力する。
        log_level: ログレベル。

    """
    if cls.sessionmaker is None:
        raise RuntimeError("init()が呼ばれていません。")
    token = cls.start_session(name=name, log_level=log_level)
    try:
        yield cls.session()
    finally:
        cls.close_session(token, name=name, log_level=log_level)

asession_scope(name=None, log_level=logging.DEBUG) async classmethod

セッションを開始するコンテキストマネージャ。

使用例:: async with Base.asession_scope() as session: ...

引数：

名前 タイプ デスクリプション デフォルト
name str | None

セッション名。指定時のみログ出力する。

 None
log_level int

ログレベル。

 DEBUG
ソースコード位置： pytilpack/sqlalchemy/sync.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
@classmethod
@contextlib.asynccontextmanager
async def asession_scope(
    cls,
    name: str | None = None,
    log_level: int = logging.DEBUG,
) -> typing.AsyncGenerator[sqlalchemy.orm.Session, None]:
    """セッションを開始するコンテキストマネージャ。

    使用例::
        async with Base.asession_scope() as session:
            ...

    Args:
        name: セッション名。指定時のみログ出力する。
        log_level: ログレベル。

    """
    if cls.sessionmaker is None:
        raise RuntimeError("init()が呼ばれていません。")
    token = cls.start_session(name=name, log_level=log_level)
    try:
        yield cls.session()
    finally:
        cls.close_session(token, name=name, log_level=log_level)

start_session(name=None, log_level=logging.DEBUG) classmethod

セッションを開始する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
@classmethod
def start_session(
    cls, name: str | None = None, log_level: int = logging.DEBUG
) -> contextvars.Token[sqlalchemy.orm.Session]:
    """セッションを開始する。"""
    if cls.sessionmaker is None:
        raise RuntimeError("init()が呼ばれていません。")
    session = cls.sessionmaker()  # pylint: disable=not-callable
    token = cls.session_var.set(session)
    if name is not None:
        logger.log(
            log_level,
            f"セッション開始: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    return token

close_session(token, name=None, log_level=logging.DEBUG) classmethod

セッションを終了する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@classmethod
def close_session(
    cls, token: contextvars.Token[sqlalchemy.orm.Session], name: str | None = None, log_level: int = logging.DEBUG
) -> None:
    """セッションを終了する。"""
    session = cls.session()
    if name is not None:
        logger.log(
            log_level,
            f"セッション終了: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    safe_close(session)
    cls.session_var.reset(token)

session() classmethod

セッションを取得する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
282
283
284
285
286
287
288
@classmethod
def session(cls) -> sqlalchemy.orm.Session:
    """セッションを取得する。"""
    sess = cls.session_var.get(None)
    if sess is None:
        raise RuntimeError(f"セッションが開始されていません。{cls.__qualname__}.start_session()を呼び出してください。")
    return sess

select() classmethod

sqlalchemy.Selectを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
290
291
292
293
294
@classmethod
def select(cls) -> sqlalchemy.Select[tuple[typing.Self]]:
    """sqlalchemy.Selectを返す。"""
    # cls.count()などでfrom句が消えないように明示的にfrom句を指定して返す。
    return sqlalchemy.select(cls).select_from(cls)

insert() classmethod

sqlalchemy.Insertを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
296
297
298
299
@classmethod
def insert(cls) -> sqlalchemy.Insert:
    """sqlalchemy.Insertを返す。"""
    return sqlalchemy.insert(cls)

update() classmethod

sqlalchemy.Updateを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
301
302
303
304
@classmethod
def update(cls) -> sqlalchemy.Update:
    """sqlalchemy.Updateを返す。"""
    return sqlalchemy.update(cls)

delete() classmethod

sqlalchemy.Deleteを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
306
307
308
309
@classmethod
def delete(cls) -> sqlalchemy.Delete:
    """sqlalchemy.Deleteを返す。"""
    return sqlalchemy.delete(cls)

count(query) classmethod

queryのレコード数を取得する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
311
312
313
314
315
316
317
@classmethod
def count(cls, query: sqlalchemy.Select | sqlalchemy.CompoundSelect) -> int:
    """queryのレコード数を取得する。"""
    # pylint: disable=not-callable
    return (
        cls.scalar_one_or_none(sqlalchemy.select(sqlalchemy.func.count()).select_from(query.order_by(None).subquery())) or 0
    )

scalar_one(query) classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
@classmethod
def scalar_one[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).scalar_one()

scalar_one_or_none(query) classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
@classmethod
def scalar_one_or_none[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).scalar_one_or_none()

scalars(query) classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[T]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
352
353
354
355
356
357
358
359
360
361
362
363
@classmethod
def scalars[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> list[T]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list(cls.session().execute(query).scalars().all())

one(query) classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT]

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
@classmethod
def one[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT]:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).one()

one_or_none(query) classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT] | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
@classmethod
def one_or_none[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT] | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).one_or_none()

all(query) classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[Row[TT]]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
398
399
400
401
402
403
404
405
406
407
408
409
@classmethod
def all[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> list[sqlalchemy.Row[TT]]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list(cls.session().execute(query).all())

get_by_id_not_null(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。見つからない場合は例外を送出する。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self

インスタンス。

Raises: ValueError: 見つからない場合。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
@classmethod
def get_by_id_not_null(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self:
    """IDを元にインスタンスを取得。見つからない場合は例外を送出する。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。
    Raises:
        ValueError: 見つからない場合。
    """
    instance = cls.get_by_id(id_, for_update=for_update, options=options)
    if instance is None:
        raise ValueError(f"{cls.__qualname__}が見つかりませんでした。id={id_}")
    return instance

get_by_id(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用。

 None

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
@classmethod
def get_by_id(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self | None:
    """IDを元にインスタンスを取得。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用。

    Returns:
        インスタンス。

    """
    q = cls.select().where(cls.id == id_)  # type: ignore  # pylint: disable=no-member
    if options is not None:
        q = q.options(options)
    if for_update:
        q = q.with_for_update()
    return cls.scalar_one_or_none(q)

paginate(query, page, per_page, scalar=True) classmethod

Flask-SQLAlchemy風ページネーション。

引数：

名前 タイプ デスクリプション デフォルト
query Select | CompoundSelect

ページネーションするクエリ。

 必須
page int

ページ番号。

 必須
per_page int

1ページあたりのアイテム数。

 必須
scalar bool

Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

 True

戻り値：

タイプ デスクリプション
Paginator

ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
@classmethod
def paginate(
    cls,
    query: sqlalchemy.Select | sqlalchemy.CompoundSelect,
    page: int,
    per_page: int,
    scalar: bool = True,
) -> pytilpack.paginator.Paginator:
    """Flask-SQLAlchemy風ページネーション。

    Args:
        query: ページネーションするクエリ。
        page: ページ番号。
        per_page: 1ページあたりのアイテム数。
        scalar: Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

    Returns:
        ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
    """
    if page <= 0:
        raise ValueError(f"ページ番号は1以上でなければなりません: page={page}")
    if per_page <= 0:
        raise ValueError(f"1ページあたりのアイテム数は1以上でなければなりません: per_page={per_page}")
    total = cls.count(query)
    page_query = query.offset((page - 1) * per_page).limit(per_page)
    items = cls.scalars(page_query) if scalar else cls.all(page_query)
    # pylint: disable=protected-access
    return pytilpack.paginator.Paginator(page=page, per_page=per_page, items=items, total=total)

commit() classmethod

セッションをコミットする。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
483
484
485
486
@classmethod
def commit(cls) -> None:
    """セッションをコミットする。"""
    cls.session().commit()

SyncUniqueIDMixin

self.unique_idを持つテーブルクラスにユニークID操作メソッドを付与するmixin。

generate_unique_id() classmethod

ユニークIDを生成する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
492
493
494
495
@classmethod
def generate_unique_id(cls) -> str:
    """ユニークIDを生成する。"""
    return secrets.token_urlsafe(32)

get_by_unique_id(unique_id, allow_id=False, for_update=False) classmethod

ユニークIDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
unique_id str | int

ユニークID。

 必須
allow_id bool

ユニークIDだけでなくID(int)も許可するかどうか。

 False
for_update bool

更新ロックを取得するか否か。

 False

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
@classmethod
def get_by_unique_id(
    cls: type[typing.Self],
    unique_id: str | int,
    allow_id: bool = False,
    for_update: bool = False,
) -> typing.Self | None:
    """ユニークIDを元にインスタンスを取得。

    Args:
        unique_id: ユニークID。
        allow_id: ユニークIDだけでなくID(int)も許可するかどうか。
        for_update: 更新ロックを取得するか否か。

    Returns:
        インスタンス。

    """
    assert issubclass(cls, SyncMixin)
    if allow_id and isinstance(unique_id, int):
        q = cls.select().where(cls.id == unique_id)  # type: ignore
    else:
        q = cls.select().where(cls.unique_id == unique_id)  # type: ignore
    if for_update:
        q = q.with_for_update()
    return cls.scalar_one_or_none(q)

asafe_close(session, log_level=logging.DEBUG) async

例外を送出せずにセッションをクローズ。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
531
532
533
534
535
536
537
async def asafe_close(session: sqlalchemy.ext.asyncio.AsyncSession, log_level: int | None = logging.DEBUG):
    """例外を送出せずにセッションをクローズ。"""
    try:
        await asyncio.shield(session.close())
    except Exception:
        if log_level is not None:
            logger.log(log_level, "セッションクローズ失敗", exc_info=True)

await_for_connection(url, timeout=180.0) async

DBに接続可能になるまで待機する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
async def await_for_connection(url: str, timeout: float = 180.0) -> None:
    """DBに接続可能になるまで待機する。"""
    state = _WaitForConnectionState(url, timeout)
    while True:
        try:
            engine = sqlalchemy.ext.asyncio.create_async_engine(url)
            try:
                async with engine.connect() as connection:
                    await connection.execute(sqlalchemy.text("SELECT 1"))
            finally:
                await engine.dispose()
            state.on_success()
            break
        except Exception as e:
            await asyncio.sleep(state.on_failure(e))

describe_table(table, orm_class, tablefmt='grid')

テーブル構造を文字列化する。

ソースコード位置： pytilpack/sqlalchemy/describe.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def describe_table(
    table: sqlalchemy.sql.schema.Table,
    orm_class: type[sqlalchemy.orm.DeclarativeBase],
    tablefmt: "str | tabulate.TableFormat" = "grid",
) -> str:
    """テーブル構造を文字列化する。"""
    try:
        class_field_comments = pytilpack.python.class_field_comments(orm_class)
    except Exception as e:
        logger.warning(f"クラスフィールドコメント取得失敗: {e}")
        class_field_comments = {}

    headers = ["Field", "Type", "Null", "Key", "Default", "Extra", "Comment"]
    rows = []
    for column in table.columns:
        key = ""
        if column.primary_key:
            key = "PRI"
        elif column.unique:
            key = "UNI"
        elif column.index:
            key = "MUL"

        extra = ""
        if column.autoincrement and column.primary_key:
            extra = "auto_increment"

        default_value = (
            column.default.arg
            if column.default is not None and isinstance(column.default, sqlalchemy.sql.schema.ColumnDefault)
            else column.default
        )
        default: str
        if default_value is None:
            default = "NULL"
        elif callable(default_value):
            default = "(function)"
        elif isinstance(default_value, sqlalchemy.sql.elements.CompilerElement):  # type: ignore[attr-defined]
            default = str(default_value.compile(compile_kwargs={"literal_binds": True}))
        else:
            default = str(default_value)

        # コメントは以下の優先順位で拾う。
        # doc(DBに反映されないもの) > comment(DBに反映されるもの)
        #  > class_field_comments(ソースコード上のコメント)
        comment: str = ""
        if column.doc:
            comment = column.doc
        elif column.comment:
            comment = column.comment
        elif column.name in class_field_comments:
            comment = class_field_comments[column.name] or ""

        rows.append(
            [
                column.name,
                str(column.type),
                "YES" if column.nullable else "NO",
                key,
                default,
                extra,
                comment,
            ]
        )
    table_description = tabulate.tabulate(rows, headers=headers, tablefmt=tablefmt)

    return f"Table: {table.name}\n{table_description}\n"

get_class_by_table(base, table)

テーブルからクラスを取得する。

ソースコード位置： pytilpack/sqlalchemy/describe.py 
27
28
29
30
31
32
33
34
def get_class_by_table(
    base: type[sqlalchemy.orm.DeclarativeBase], table: sqlalchemy.sql.schema.Table
) -> type[sqlalchemy.orm.DeclarativeBase]:
    """テーブルからクラスを取得する。"""
    for mapper in base.registry.mappers:
        if mapper.local_table is table:
            return typing.cast(type[sqlalchemy.orm.DeclarativeBase], mapper.class_)
    raise ValueError(f"テーブル {table.name} に対応するクラスが見つかりませんでした。")

register_ping()

コネクションプールの切断対策。

ソースコード位置： pytilpack/sqlalchemy/flask.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def register_ping():
    """コネクションプールの切断対策。"""

    @sqlalchemy.event.listens_for(sqlalchemy.pool.Pool, "checkout")
    def _ping_connection(dbapi_connection, connection_record, connection_proxy):
        """コネクションプールの切断対策。"""
        _ = connection_record, connection_proxy  # noqa
        cursor = dbapi_connection.cursor()
        try:
            cursor.execute("SELECT 1")
        except Exception as e:
            raise sqlalchemy.exc.DisconnectionError() from e
        finally:
            cursor.close()

run_sync_with_session(func)

同期関数を非同期に実行し、スレッド内でセッションを管理するデコレーター。

別スレッドで実行される関数内で自動的にセッションスコープを作成する。 各呼び出しは独立したセッション・トランザクションを持つ。

引数：

名前 タイプ デスクリプション デフォルト
func Callable[Concatenate[type[SyncMixin], P], R]

デコレート対象の同期関数

 必須

戻り値：

タイプ デスクリプション
Callable[Concatenate[type[SyncMixin], P], Awaitable[R]]

非同期版の関数
ソースコード位置： pytilpack/sqlalchemy/sync.py 
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
def run_sync_with_session[**P, R](
    func: "typing.Callable[typing.Concatenate[type[SyncMixin], P], R]",
) -> "typing.Callable[typing.Concatenate[type[SyncMixin], P], typing.Awaitable[R]]":
    """同期関数を非同期に実行し、スレッド内でセッションを管理するデコレーター。

    別スレッドで実行される関数内で自動的にセッションスコープを作成する。
    各呼び出しは独立したセッション・トランザクションを持つ。

    Args:
        func: デコレート対象の同期関数

    Returns:
        非同期版の関数
    """

    @functools.wraps(func)
    async def wrapper(cls: type[SyncMixin], *args: P.args, **kwargs: P.kwargs) -> R:
        def _impl() -> R:
            with cls.session_scope() as session:
                result = func(cls, *args, **kwargs)
                session.commit()
                return result

        return await asyncio.to_thread(_impl)

    return wrapper

safe_close(session, log_level=logging.DEBUG)

例外を送出せずにセッションをクローズ。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
543
544
545
546
547
548
549
550
551
552
def safe_close(
    session: sqlalchemy.orm.Session | sqlalchemy.orm.scoped_session,
    log_level: int | None = logging.DEBUG,
):
    """例外を送出せずにセッションをクローズ。"""
    try:
        session.close()
    except Exception:
        if log_level is not None:
            logger.log(log_level, "セッションクローズ失敗", exc_info=True)

wait_for_connection(url, timeout=180.0)

DBに接続可能になるまで待機する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
def wait_for_connection(url: str, timeout: float = 180.0) -> None:
    """DBに接続可能になるまで待機する。"""
    state = _WaitForConnectionState(url, timeout)
    while True:
        try:
            engine = sqlalchemy.create_engine(url)
            try:
                with engine.connect() as connection:
                    result = connection.execute(sqlalchemy.text("SELECT 1"))
                    result.close()
            finally:
                engine.dispose()
            state.on_success()
            break
        except Exception as e:
            time.sleep(state.on_failure(e))

async_

SQLAlchemy用のユーティリティ集（async版）。

AsyncMixin

Bases: AsyncAttrs, _ReprMixin, _ToDictMixin

モデルのベースクラス。SQLAlchemy 2.0スタイル・async前提。

例：

モデル定義例::

class Base(sqlalchemy.orm.DeclarativeBase, pytilpack.sqlalchemy_.AsyncMixin):
    pass

class User(Base):
    __tablename__ = "users"
    ...

Quart例::

@app.before_request
async def _before_request() -> None:
    quart.g.db_session_token = await models.Base.start_session()

@app.teardown_request
async def _teardown_request(_: BaseException | None) -> None:
    if hasattr(quart.g, "db_session_token"):
        await models.Base.close_session(quart.g.db_session_token)
        del quart.g.db_session_token
session_var = contextvars.ContextVar('session_var') class-attribute instance-attribute

セッション。

init(url, pool_size=None, max_overflow=None, pool_recycle=280, pool_pre_ping=True, autoflush=True, expire_on_commit=False, eagerly_init=True, **kwargs) classmethod

DB接続を初期化する。

engineとsessionmakerはスレッドローカルで遅延初期化される。 各スレッドが初めてDB操作を行う際にそのスレッド専用のengineとsessionmakerが生成される。

引数：

名前 タイプ デスクリプション デフォルト
url str | URL

DB接続URL。

 必須
pool_size int | None

コネクションプールのサイズ。スレッド数に応じて調整要。負数の場合はプーリングを無効化する。

 None
max_overflow int | None

コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。

 None
pool_recycle int | None

コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。

 280
pool_pre_ping bool

コネクションプールのプレピン。Noneの場合はデフォルト値を使用。

 True
autoflush bool

セッションのautoflushフラグ。デフォルトはTrue。

 True
expire_on_commit bool

セッションのexpire_on_commitフラグ。デフォルトはFalse。

 False
eagerly_init bool

Trueの場合（デフォルト）、呼び出し元スレッドのengineとsessionmakerをすぐに生成する。

 True
**kwargs Any

その他のsqlalchemy.create_async_engine()へのキーワード引数。

 {}
ソースコード位置： pytilpack/sqlalchemy/async_.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@classmethod
def init(
    cls,
    url: str | sqlalchemy.engine.URL,
    pool_size: int | None = None,
    max_overflow: int | None = None,
    pool_recycle: int | None = 280,
    pool_pre_ping: bool = True,
    autoflush: bool = True,
    expire_on_commit: bool = False,
    eagerly_init: bool = True,
    **kwargs: typing.Any,
) -> None:
    """DB接続を初期化する。

    engineとsessionmakerはスレッドローカルで遅延初期化される。
    各スレッドが初めてDB操作を行う際にそのスレッド専用のengineとsessionmakerが生成される。

    Args:
        url: DB接続URL。
        pool_size: コネクションプールのサイズ。スレッド数に応じて調整要。負数の場合はプーリングを無効化する。
        max_overflow: コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。
        pool_recycle: コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。
        pool_pre_ping: コネクションプールのプレピン。Noneの場合はデフォルト値を使用。
        autoflush: セッションのautoflushフラグ。デフォルトはTrue。
        expire_on_commit: セッションのexpire_on_commitフラグ。デフォルトはFalse。
        eagerly_init: Trueの場合（デフォルト）、呼び出し元スレッドのengineとsessionmakerをすぐに生成する。
        **kwargs: その他のsqlalchemy.create_async_engine()へのキーワード引数。

    """
    if cls._init_args is not None:
        raise RuntimeError("DB接続はすでに初期化されています。")

    if pool_size is not None and pool_size < 0:
        engine_kwargs = kwargs.copy()
        engine_kwargs["poolclass"] = sqlalchemy.pool.NullPool
    else:
        if pool_size is not None and max_overflow is None:
            max_overflow = pool_size * 1  # デフォルトで倍まで許可
        engine_kwargs = kwargs.copy()
        if pool_size is not None:
            engine_kwargs["pool_size"] = pool_size
        if max_overflow is not None:
            engine_kwargs["max_overflow"] = max_overflow
    if pool_recycle is not None:
        engine_kwargs["pool_recycle"] = pool_recycle
    if pool_pre_ping is not None:
        engine_kwargs["pool_pre_ping"] = pool_pre_ping

    cls._init_args = _InitArgs(
        url=url,
        autoflush=autoflush,
        expire_on_commit=expire_on_commit,
        engine_kwargs=engine_kwargs,
    )
    cls._thread_local = threading.local()

    if eagerly_init:
        cls._get_engine_and_sessionmaker()
term() async classmethod

現在のスレッドのDB接続を終了する。

スレッドローカルなengineをdisposeして解放する。 スレッド終了時やテスト後の後始末に使用する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@classmethod
async def term(cls) -> None:
    """現在のスレッドのDB接続を終了する。

    スレッドローカルなengineをdisposeして解放する。
    スレッド終了時やテスト後の後始末に使用する。

    """
    if cls._init_args is None:
        logger.warning("未初期化時のterm()呼び出し")
        return
    if cls._thread_local is None:
        return
    if hasattr(cls._thread_local, "engine") and cls._thread_local.engine is not None:
        await cls._thread_local.engine.dispose()
        cls._thread_local.engine = None
        cls._thread_local.sessionmaker = None
engine() classmethod

現在のスレッドのDB接続を返す。init()後に初めて呼ばれた時点で遅延生成される。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
170
171
172
173
@classmethod
def engine(cls) -> sqlalchemy.ext.asyncio.AsyncEngine:
    """現在のスレッドのDB接続を返す。init()後に初めて呼ばれた時点で遅延生成される。"""
    return cls._get_engine_and_sessionmaker()[0]
sessionmaker() classmethod

現在のスレッドのセッションファクトリを返す。init()後に初めて呼ばれた時点で遅延生成される。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
175
176
177
178
179
180
@classmethod
def sessionmaker(
    cls,
) -> sqlalchemy.ext.asyncio.async_sessionmaker[sqlalchemy.ext.asyncio.AsyncSession]:
    """現在のスレッドのセッションファクトリを返す。init()後に初めて呼ばれた時点で遅延生成される。"""
    return cls._get_engine_and_sessionmaker()[1]
connect() classmethod

DBに接続する。

使用例:: async with Base.connect() as conn: await conn.run_sync(Base.metadata.create_all)

ソースコード位置： pytilpack/sqlalchemy/async_.py 
182
183
184
185
186
187
188
189
190
191
@classmethod
def connect(cls) -> sqlalchemy.ext.asyncio.AsyncConnection:
    """DBに接続する。

    使用例::
        async with Base.connect() as conn:
            await conn.run_sync(Base.metadata.create_all)

    """
    return cls.engine().connect()
session_scope(name=None, log_level=logging.DEBUG) async classmethod

セッションを開始するコンテキストマネージャ。

使用例:: async with Base.session_scope() as session: ...

引数：

名前 タイプ デスクリプション デフォルト
name str | None

セッション名。指定時のみログ出力する。

 None
log_level int

ログレベル。

 DEBUG
ソースコード位置： pytilpack/sqlalchemy/async_.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@classmethod
@contextlib.asynccontextmanager
async def session_scope(
    cls,
    name: str | None = None,
    log_level: int = logging.DEBUG,
):
    """セッションを開始するコンテキストマネージャ。

    使用例::
        async with Base.session_scope() as session:
            ...

    Args:
        name: セッション名。指定時のみログ出力する。
        log_level: ログレベル。

    """
    token = await cls.start_session(name=name, log_level=log_level)
    try:
        yield cls.session()
    finally:
        await cls.close_session(token, name=name, log_level=log_level)
start_session(name=None, log_level=logging.DEBUG) async classmethod

セッションを開始する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@classmethod
async def start_session(
    cls, name: str | None = None, log_level: int = logging.DEBUG
) -> contextvars.Token[sqlalchemy.ext.asyncio.AsyncSession]:
    """セッションを開始する。"""
    session = cls.sessionmaker()()  # pylint: disable=not-callable
    token = cls.session_var.set(session)
    if name is not None:
        logger.log(
            log_level,
            f"セッション開始: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    return token
close_session(token, name=None, log_level=logging.DEBUG) async classmethod

セッションを終了する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@classmethod
async def close_session(
    cls,
    token: contextvars.Token[sqlalchemy.ext.asyncio.AsyncSession],
    name: str | None = None,
    log_level: int = logging.DEBUG,
) -> None:
    """セッションを終了する。"""
    session = cls.session()
    if name is not None:
        logger.log(
            log_level,
            f"セッション終了: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    await asafe_close(session)
    cls.session_var.reset(token)
session() classmethod

セッションを取得する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
252
253
254
255
256
257
258
@classmethod
def session(cls) -> sqlalchemy.ext.asyncio.AsyncSession:
    """セッションを取得する。"""
    sess = cls.session_var.get(None)
    if sess is None:
        raise RuntimeError(f"セッションが開始されていません。{cls.__qualname__}.start_session()を呼び出してください。")
    return sess
select() classmethod

sqlalchemy.Selectを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
260
261
262
263
264
@classmethod
def select(cls) -> sqlalchemy.Select[tuple[typing.Self]]:
    """sqlalchemy.Selectを返す。"""
    # cls.count()などでfrom句が消えないように明示的にfrom句を指定して返す。
    return sqlalchemy.select(cls).select_from(cls)
insert() classmethod

sqlalchemy.Insertを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
266
267
268
269
@classmethod
def insert(cls) -> sqlalchemy.Insert:
    """sqlalchemy.Insertを返す。"""
    return sqlalchemy.insert(cls)
update() classmethod

sqlalchemy.Updateを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
271
272
273
274
@classmethod
def update(cls) -> sqlalchemy.Update:
    """sqlalchemy.Updateを返す。"""
    return sqlalchemy.update(cls)
delete() classmethod

sqlalchemy.Deleteを返す。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
276
277
278
279
@classmethod
def delete(cls) -> sqlalchemy.Delete:
    """sqlalchemy.Deleteを返す。"""
    return sqlalchemy.delete(cls)
count(query) async classmethod

queryのレコード数を取得する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
281
282
283
284
285
286
287
288
289
290
@classmethod
async def count(cls, query: sqlalchemy.Select | sqlalchemy.CompoundSelect) -> int:
    """queryのレコード数を取得する。"""
    # pylint: disable=not-callable
    return (
        await cls.scalar_one_or_none(
            sqlalchemy.select(sqlalchemy.func.count()).select_from(query.order_by(None).subquery())
        )
        or 0
    )
scalar_one(query) async classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@classmethod
async def scalar_one[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).scalar_one()
scalar_one_or_none(query) async classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
@classmethod
async def scalar_one_or_none[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).scalar_one_or_none()
scalars(query) async classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[T]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
325
326
327
328
329
330
331
332
333
334
335
336
@classmethod
async def scalars[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> list[T]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list((await cls.session().execute(query)).scalars().all())
one(query) async classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT]

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
@classmethod
async def one[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT]:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).one()
one_or_none(query) async classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT] | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
@classmethod
async def one_or_none[TT: tuple](
    cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]
) -> sqlalchemy.Row[TT] | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return (await cls.session().execute(query)).one_or_none()
all(query) async classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[Row[TT]]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
373
374
375
376
377
378
379
380
381
382
383
384
@classmethod
async def all[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> list[sqlalchemy.Row[TT]]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list((await cls.session().execute(query)).all())
get_by_id_not_null(id_, for_update=False, options=None) async classmethod

IDを元にインスタンスを取得。見つからない場合は例外を送出する。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self

インスタンス。

Raises: ValueError: 見つからない場合。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
@classmethod
async def get_by_id_not_null(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self:
    """IDを元にインスタンスを取得。見つからない場合は例外を送出する。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。
    Raises:
        ValueError: 見つからない場合。
    """
    instance = await cls.get_by_id(id_, for_update=for_update, options=options)
    if instance is None:
        raise ValueError(f"{cls.__qualname__}が見つかりませんでした。id={id_}")
    return instance
get_by_id(id_, for_update=False, options=None) async classmethod

IDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
@classmethod
async def get_by_id(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self | None:
    """IDを元にインスタンスを取得。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。

    """
    q = cls.select().where(cls.id == id_)  # type: ignore  # pylint: disable=no-member
    if options is not None:
        q = q.options(options)
    if for_update:
        q = q.with_for_update()
    return await cls.scalar_one_or_none(q)  # type: ignore[arg-type]
paginate(query, page, per_page, scalar=True) async classmethod

Flask-SQLAlchemy風ページネーション。

引数：

名前 タイプ デスクリプション デフォルト
query Select | CompoundSelect

ページネーションするクエリ。

 必須
page int

ページ番号。

 必須
per_page int

1ページあたりのアイテム数。

 必須
scalar bool

Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

 True

戻り値：

タイプ デスクリプション
Paginator

ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
@classmethod
async def paginate(
    cls,
    query: sqlalchemy.Select | sqlalchemy.CompoundSelect,
    page: int,
    per_page: int,
    scalar: bool = True,
) -> pytilpack.paginator.Paginator:
    """Flask-SQLAlchemy風ページネーション。

    Args:
        query: ページネーションするクエリ。
        page: ページ番号。
        per_page: 1ページあたりのアイテム数。
        scalar: Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

    Returns:
        ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
    """
    if page <= 0:
        raise ValueError(f"ページ番号は1以上でなければなりません: page={page}")
    if per_page <= 0:
        raise ValueError(f"1ページあたりのアイテム数は1以上でなければなりません: per_page={per_page}")
    total = await cls.count(query)
    page_query = query.offset((page - 1) * per_page).limit(per_page)
    items = await (cls.scalars(page_query) if scalar else cls.all(page_query))
    # pylint: disable=protected-access
    return pytilpack.paginator.Paginator(page=page, per_page=per_page, items=items, total=total)
run_with_session(func, *args, **kwargs) classmethod

非同期関数をセッション付きで同期実行する関数。

引数：

名前 タイプ デスクリプション デフォルト
func Callable[P, Awaitable[R]]

デコレート対象の非同期関数

 必須
*args P.args

非同期関数への引数

 ()
**kwargs P.kwargs

非同期関数へのキーワード引数

 {}

戻り値：

タイプ デスクリプション
R

非同期関数の戻り値
ソースコード位置： pytilpack/sqlalchemy/async_.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
@classmethod
def run_with_session[**P, R](cls, func: typing.Callable[P, typing.Awaitable[R]], *args: P.args, **kwargs: P.kwargs) -> R:
    """非同期関数をセッション付きで同期実行する関数。

    Args:
        func: デコレート対象の非同期関数
        *args: 非同期関数への引数
        **kwargs: 非同期関数へのキーワード引数

    Returns:
        非同期関数の戻り値
    """

    async def wrapper() -> R:
        async with cls.session_scope():
            return await func(*args, **kwargs)

    return pytilpack.asyncio.run(wrapper())

AsyncUniqueIDMixin

self.unique_idを持つテーブルクラスにユニークID操作メソッドを付与するmixin。

generate_unique_id() classmethod

ユニークIDを生成する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
481
482
483
484
@classmethod
def generate_unique_id(cls) -> str:
    """ユニークIDを生成する。"""
    return secrets.token_urlsafe(32)
get_by_unique_id(unique_id, allow_id=False, for_update=False) async classmethod

ユニークIDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
unique_id str | int

ユニークID。

 必須
allow_id bool

ユニークIDだけでなくID(int)も許可するかどうか。

 False
for_update bool

更新ロックを取得するか否か。

 False

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/async_.py 
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
@classmethod
async def get_by_unique_id(
    cls: type[typing.Self],
    unique_id: str | int,
    allow_id: bool = False,
    for_update: bool = False,
) -> typing.Self | None:
    """ユニークIDを元にインスタンスを取得。

    Args:
        unique_id: ユニークID。
        allow_id: ユニークIDだけでなくID(int)も許可するかどうか。
        for_update: 更新ロックを取得するか否か。

    Returns:
        インスタンス。

    """
    assert issubclass(cls, AsyncMixin)
    if allow_id and isinstance(unique_id, int):
        q = cls.select().where(cls.id == unique_id)  # type: ignore
    else:
        q = cls.select().where(cls.unique_id == unique_id)  # type: ignore
    if for_update:
        q = q.with_for_update()
    return await cls.scalar_one_or_none(q)

await_for_connection(url, timeout=180.0) async

DBに接続可能になるまで待機する。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
async def await_for_connection(url: str, timeout: float = 180.0) -> None:
    """DBに接続可能になるまで待機する。"""
    state = _WaitForConnectionState(url, timeout)
    while True:
        try:
            engine = sqlalchemy.ext.asyncio.create_async_engine(url)
            try:
                async with engine.connect() as connection:
                    await connection.execute(sqlalchemy.text("SELECT 1"))
            finally:
                await engine.dispose()
            state.on_success()
            break
        except Exception as e:
            await asyncio.sleep(state.on_failure(e))

asafe_close(session, log_level=logging.DEBUG) async

例外を送出せずにセッションをクローズ。

ソースコード位置： pytilpack/sqlalchemy/async_.py 
531
532
533
534
535
536
537
async def asafe_close(session: sqlalchemy.ext.asyncio.AsyncSession, log_level: int | None = logging.DEBUG):
    """例外を送出せずにセッションをクローズ。"""
    try:
        await asyncio.shield(session.close())
    except Exception:
        if log_level is not None:
            logger.log(log_level, "セッションクローズ失敗", exc_info=True)

describe

SQLAlchemy用のユーティリティ集。

describe(Base, tablefmt='grid')

DBのテーブル構造を文字列化する。

ソースコード位置： pytilpack/sqlalchemy/describe.py 
17
18
19
20
21
22
23
24
def describe(
    Base: type[sqlalchemy.orm.DeclarativeBase],
    tablefmt: "str | tabulate.TableFormat" = "grid",
) -> str:
    """DBのテーブル構造を文字列化する。"""
    return "\n".join(
        [describe_table(table, get_class_by_table(Base, table), tablefmt=tablefmt) for table in Base.metadata.tables.values()]
    )

get_class_by_table(base, table)

テーブルからクラスを取得する。

ソースコード位置： pytilpack/sqlalchemy/describe.py 
27
28
29
30
31
32
33
34
def get_class_by_table(
    base: type[sqlalchemy.orm.DeclarativeBase], table: sqlalchemy.sql.schema.Table
) -> type[sqlalchemy.orm.DeclarativeBase]:
    """テーブルからクラスを取得する。"""
    for mapper in base.registry.mappers:
        if mapper.local_table is table:
            return typing.cast(type[sqlalchemy.orm.DeclarativeBase], mapper.class_)
    raise ValueError(f"テーブル {table.name} に対応するクラスが見つかりませんでした。")

describe_table(table, orm_class, tablefmt='grid')

テーブル構造を文字列化する。

ソースコード位置： pytilpack/sqlalchemy/describe.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def describe_table(
    table: sqlalchemy.sql.schema.Table,
    orm_class: type[sqlalchemy.orm.DeclarativeBase],
    tablefmt: "str | tabulate.TableFormat" = "grid",
) -> str:
    """テーブル構造を文字列化する。"""
    try:
        class_field_comments = pytilpack.python.class_field_comments(orm_class)
    except Exception as e:
        logger.warning(f"クラスフィールドコメント取得失敗: {e}")
        class_field_comments = {}

    headers = ["Field", "Type", "Null", "Key", "Default", "Extra", "Comment"]
    rows = []
    for column in table.columns:
        key = ""
        if column.primary_key:
            key = "PRI"
        elif column.unique:
            key = "UNI"
        elif column.index:
            key = "MUL"

        extra = ""
        if column.autoincrement and column.primary_key:
            extra = "auto_increment"

        default_value = (
            column.default.arg
            if column.default is not None and isinstance(column.default, sqlalchemy.sql.schema.ColumnDefault)
            else column.default
        )
        default: str
        if default_value is None:
            default = "NULL"
        elif callable(default_value):
            default = "(function)"
        elif isinstance(default_value, sqlalchemy.sql.elements.CompilerElement):  # type: ignore[attr-defined]
            default = str(default_value.compile(compile_kwargs={"literal_binds": True}))
        else:
            default = str(default_value)

        # コメントは以下の優先順位で拾う。
        # doc(DBに反映されないもの) > comment(DBに反映されるもの)
        #  > class_field_comments(ソースコード上のコメント)
        comment: str = ""
        if column.doc:
            comment = column.doc
        elif column.comment:
            comment = column.comment
        elif column.name in class_field_comments:
            comment = class_field_comments[column.name] or ""

        rows.append(
            [
                column.name,
                str(column.type),
                "YES" if column.nullable else "NO",
                key,
                default,
                extra,
                comment,
            ]
        )
    table_description = tabulate.tabulate(rows, headers=headers, tablefmt=tablefmt)

    return f"Table: {table.name}\n{table_description}\n"

flask

SQLAlchemy用のユーティリティ集（Flask-SQLAlchemy版）。

Mixin

Bases: _ReprMixin, _ToDictMixin

テーブルクラスに共通メソッドを付与するMixin。

get_by_id_not_null(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。見つからない場合は例外を送出する。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self

インスタンス。

Raises: ValueError: 見つからない場合。

ソースコード位置： pytilpack/sqlalchemy/flask.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
def get_by_id_not_null(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self:
    """IDを元にインスタンスを取得。見つからない場合は例外を送出する。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。
    Raises:
        ValueError: 見つからない場合。
    """
    instance = cls.get_by_id(id_, for_update=for_update, options=options)
    if instance is None:
        raise ValueError(f"{cls.__qualname__}が見つかりませんでした。id={id_}")
    return instance
get_by_id(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用。

 None

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/flask.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@classmethod
def get_by_id(
    cls: type[typing.Self], id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self | None:
    """IDを元にインスタンスを取得。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用。

    Returns:
        インスタンス。

    """
    q = cls.query.filter(cls.id == id_)  # type: ignore
    if options is not None:
        q = q.options(options)
    if for_update:
        q = q.with_for_update()
    return q.one_or_none()

UniqueIDMixin

self.unique_idを持つテーブルクラスにユニークID操作メソッドを付与するmixin。

generate_unique_id() classmethod

ユニークIDを生成する。

ソースコード位置： pytilpack/sqlalchemy/flask.py 
84
85
86
87
@classmethod
def generate_unique_id(cls) -> str:
    """ユニークIDを生成する。"""
    return secrets.token_urlsafe(32)
get_by_unique_id(unique_id, allow_id=False, for_update=False) classmethod

ユニークIDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
unique_id str | int

ユニークID。

 必須
allow_id bool

ユニークIDだけでなくID(int)も許可するかどうか。

 False
for_update bool

更新ロックを取得するか否か。

 False

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/flask.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_by_unique_id(
    cls: type[typing.Self],
    unique_id: str | int,
    allow_id: bool = False,
    for_update: bool = False,
) -> typing.Self | None:
    """ユニークIDを元にインスタンスを取得。

    Args:
        unique_id: ユニークID。
        allow_id: ユニークIDだけでなくID(int)も許可するかどうか。
        for_update: 更新ロックを取得するか否か。

    Returns:
        インスタンス。

    """
    if allow_id and isinstance(unique_id, int):
        q = cls.query.filter(cls.id == unique_id)  # type: ignore
    else:
        q = cls.query.filter(cls.unique_id == unique_id)  # type: ignore
    if for_update:
        q = q.with_for_update()
    return q.one_or_none()

register_ping()

コネクションプールの切断対策。

ソースコード位置： pytilpack/sqlalchemy/flask.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def register_ping():
    """コネクションプールの切断対策。"""

    @sqlalchemy.event.listens_for(sqlalchemy.pool.Pool, "checkout")
    def _ping_connection(dbapi_connection, connection_record, connection_proxy):
        """コネクションプールの切断対策。"""
        _ = connection_record, connection_proxy  # noqa
        cursor = dbapi_connection.cursor()
        try:
            cursor.execute("SELECT 1")
        except Exception as e:
            raise sqlalchemy.exc.DisconnectionError() from e
        finally:
            cursor.close()

sync

SQLAlchemy用のユーティリティ集（同期版）。

SyncAsyncBridgeMixin

同期DBを非同期環境から扱うためのブリッジMixin。

SQLAlchemyの同期APIによるI/Oをasyncio.to_threadでスレッドに逃がし、 awaitableなインターフェースを提供する。各呼び出しは別スレッド・独立セッション ・独立トランザクションで実行される。

SyncMixinが継承する形で提供されるため、SyncMixin.acount(...)等の 呼び出しはそのまま維持される。非同期専用のコードから同期DBを利用する場合の 責務分離を明確にする目的で、別Mixinとして切り出している。

acount(query) async classmethod

queryのレコード数を返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
38
39
40
41
42
43
@classmethod
async def acount(cls, query: sqlalchemy.Select | sqlalchemy.CompoundSelect) -> int:
    """queryのレコード数を返す。非同期版。"""
    # cls.count.__func__はSyncMixin側で定義されるためmypy向けに抑制する。
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.count.__func__)(sync_cls, query)  # type: ignore[attr-defined]
ascalar_one(query) async classmethod

queryの結果を1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
45
46
47
48
49
@classmethod
async def ascalar_one[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T:
    """queryの結果を1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.scalar_one.__func__)(sync_cls, query)  # type: ignore[attr-defined]
ascalar_one_or_none(query) async classmethod

queryの結果を0件または1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
51
52
53
54
55
@classmethod
async def ascalar_one_or_none[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T | None:
    """queryの結果を0件または1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.scalar_one_or_none.__func__)(sync_cls, query)  # type: ignore[attr-defined]
ascalars(query) async classmethod

queryの結果を全件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
57
58
59
60
61
@classmethod
async def ascalars[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> list[T]:
    """queryの結果を全件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.scalars.__func__)(sync_cls, query)  # type: ignore[attr-defined]
aone(query) async classmethod

queryの結果を1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
63
64
65
66
67
@classmethod
async def aone[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT]:
    """queryの結果を1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.one.__func__)(sync_cls, query)  # type: ignore[attr-defined]
aone_or_none(query) async classmethod

queryの結果を0件または1件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
69
70
71
72
73
74
75
@classmethod
async def aone_or_none[TT: tuple](
    cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]
) -> sqlalchemy.Row[TT] | None:
    """queryの結果を0件または1件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.one_or_none.__func__)(sync_cls, query)  # type: ignore[attr-defined]
aall(query) async classmethod

queryの結果を全件返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
77
78
79
80
81
@classmethod
async def aall[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> list[sqlalchemy.Row[TT]]:
    """queryの結果を全件返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.all.__func__)(sync_cls, query)  # type: ignore[attr-defined]
aget_by_id(id_, for_update=False) async classmethod

IDを元にインスタンスを取得する。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
83
84
85
86
87
@classmethod
async def aget_by_id(cls, id_: int, for_update: bool = False) -> typing.Self | None:
    """IDを元にインスタンスを取得する。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.get_by_id.__func__)(sync_cls, id_, for_update)  # type: ignore[attr-defined,return-value]
apaginate(query, page, per_page, scalar=True) async classmethod

Flask-SQLAlchemy風ページネーションを返す。非同期版。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
89
90
91
92
93
94
95
96
97
98
99
@classmethod
async def apaginate(
    cls,
    query: sqlalchemy.Select | sqlalchemy.CompoundSelect,
    page: int,
    per_page: int,
    scalar: bool = True,
) -> pytilpack.paginator.Paginator:
    """Flask-SQLAlchemy風ページネーションを返す。非同期版。"""
    sync_cls = typing.cast("type[SyncMixin]", cls)
    return await run_sync_with_session(sync_cls.paginate.__func__)(sync_cls, query, page, per_page, scalar)  # type: ignore[attr-defined]

SyncMixin

Bases: _ReprMixin, _ToDictMixin, SyncAsyncBridgeMixin

モデルのベースクラス。SQLAlchemy 2.0スタイル・同期前提。

例：

モデル定義例::

class Base(sqlalchemy.orm.DeclarativeBase, pytilpack.sqlalchemy_.sync_.SyncMixin):
    pass

class User(Base):
    __tablename__ = "users"
    ...

Quart例::

@app.before_request
async def _before_request() -> None:
    quart.g.db_session_token = models.Base.start_session()

@app.teardown_request
async def _teardown_request(_: BaseException | None) -> None:
    if hasattr(quart.g, "db_session_token"):
        models.Base.close_session(quart.g.db_session_token)
        del quart.g.db_session_token
engine = None class-attribute instance-attribute

DB接続。

sessionmaker = None class-attribute instance-attribute

セッションファクトリ。

session_var = contextvars.ContextVar('session_var') class-attribute instance-attribute

セッション。

init(url, pool_size=None, max_overflow=None, pool_recycle=280, pool_pre_ping=True, autoflush=True, expire_on_commit=False, **kwargs) classmethod

DB接続を初期化する。

引数：

名前 タイプ デスクリプション デフォルト
url str | URL

DB接続URL。

 必須
pool_size int | None

コネクションプールのサイズ。スレッド数に応じて調整要。

 None
max_overflow int | None

コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。

 None
pool_recycle int | None

コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。

 280
pool_pre_ping bool

コネクションプールのプレピン。Noneの場合はデフォルト値を使用。

 True
autoflush bool

セッションのautoflushフラグ。デフォルトはTrue。

 True
expire_on_commit bool

セッションのexpire_on_commitフラグ。デフォルトはFalse。

 False
**kwargs Any

sqlalchemy.create_engineに渡す追加のキーワード引数。

 {}
ソースコード位置： pytilpack/sqlalchemy/sync.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@classmethod
def init(
    cls,
    url: str | sqlalchemy.engine.URL,
    pool_size: int | None = None,
    max_overflow: int | None = None,
    pool_recycle: int | None = 280,
    pool_pre_ping: bool = True,
    autoflush: bool = True,
    expire_on_commit: bool = False,
    **kwargs: typing.Any,
) -> None:
    """DB接続を初期化する。

    Args:
        url: DB接続URL。
        pool_size: コネクションプールのサイズ。スレッド数に応じて調整要。
        max_overflow: コネクションプールの最大オーバーフロー数。Noneの場合はデフォルト値を使用。
        pool_recycle: コネクションプールのリサイクル時間。Noneの場合はデフォルト値を使用。
        pool_pre_ping: コネクションプールのプレピン。Noneの場合はデフォルト値を使用。
        autoflush: セッションのautoflushフラグ。デフォルトはTrue。
        expire_on_commit: セッションのexpire_on_commitフラグ。デフォルトはFalse。
        **kwargs: sqlalchemy.create_engineに渡す追加のキーワード引数。

    """
    if cls.engine is not None:
        raise RuntimeError("DB接続はすでに初期化されています。")

    if pool_size is not None and max_overflow is None:
        max_overflow = pool_size * 1  # デフォルトで倍まで許可
    kwargs = kwargs.copy()
    if pool_size is not None:
        kwargs["pool_size"] = pool_size
    if max_overflow is not None:
        kwargs["max_overflow"] = max_overflow
    if pool_recycle is not None:
        kwargs["pool_recycle"] = pool_recycle
    if pool_pre_ping is not None:
        kwargs["pool_pre_ping"] = pool_pre_ping

    cls.engine = sqlalchemy.create_engine(url, **kwargs)
    atexit.register(cls.engine.dispose)

    cls.sessionmaker = sqlalchemy.orm.sessionmaker(cls.engine, autoflush=autoflush, expire_on_commit=expire_on_commit)
connect() classmethod

DBに接続する。

使用例:: with Base.connect() as conn: Base.metadata.create_all(conn)

ソースコード位置： pytilpack/sqlalchemy/sync.py 
183
184
185
186
187
188
189
190
191
192
193
194
@classmethod
def connect(cls) -> sqlalchemy.Connection:
    """DBに接続する。

    使用例::
        with Base.connect() as conn:
            Base.metadata.create_all(conn)

    """
    if cls.engine is None:
        raise RuntimeError("init()が呼ばれていません。")
    return cls.engine.connect()
session_scope(name=None, log_level=logging.DEBUG) classmethod

セッションを開始するコンテキストマネージャ。

使用例:: with Base.session_scope() as session: ...

引数：

名前 タイプ デスクリプション デフォルト
name str | None

セッション名。指定時のみログ出力する。

 None
log_level int

ログレベル。

 DEBUG
ソースコード位置： pytilpack/sqlalchemy/sync.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@classmethod
@contextlib.contextmanager
def session_scope(
    cls,
    name: str | None = None,
    log_level: int = logging.DEBUG,
) -> typing.Generator[sqlalchemy.orm.Session, None, None]:
    """セッションを開始するコンテキストマネージャ。

    使用例::
        with Base.session_scope() as session:
            ...

    Args:
        name: セッション名。指定時のみログ出力する。
        log_level: ログレベル。

    """
    if cls.sessionmaker is None:
        raise RuntimeError("init()が呼ばれていません。")
    token = cls.start_session(name=name, log_level=log_level)
    try:
        yield cls.session()
    finally:
        cls.close_session(token, name=name, log_level=log_level)
asession_scope(name=None, log_level=logging.DEBUG) async classmethod

セッションを開始するコンテキストマネージャ。

使用例:: async with Base.asession_scope() as session: ...

引数：

名前 タイプ デスクリプション デフォルト
name str | None

セッション名。指定時のみログ出力する。

 None
log_level int

ログレベル。

 DEBUG
ソースコード位置： pytilpack/sqlalchemy/sync.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
@classmethod
@contextlib.asynccontextmanager
async def asession_scope(
    cls,
    name: str | None = None,
    log_level: int = logging.DEBUG,
) -> typing.AsyncGenerator[sqlalchemy.orm.Session, None]:
    """セッションを開始するコンテキストマネージャ。

    使用例::
        async with Base.asession_scope() as session:
            ...

    Args:
        name: セッション名。指定時のみログ出力する。
        log_level: ログレベル。

    """
    if cls.sessionmaker is None:
        raise RuntimeError("init()が呼ばれていません。")
    token = cls.start_session(name=name, log_level=log_level)
    try:
        yield cls.session()
    finally:
        cls.close_session(token, name=name, log_level=log_level)
start_session(name=None, log_level=logging.DEBUG) classmethod

セッションを開始する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
@classmethod
def start_session(
    cls, name: str | None = None, log_level: int = logging.DEBUG
) -> contextvars.Token[sqlalchemy.orm.Session]:
    """セッションを開始する。"""
    if cls.sessionmaker is None:
        raise RuntimeError("init()が呼ばれていません。")
    session = cls.sessionmaker()  # pylint: disable=not-callable
    token = cls.session_var.set(session)
    if name is not None:
        logger.log(
            log_level,
            f"セッション開始: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    return token
close_session(token, name=None, log_level=logging.DEBUG) classmethod

セッションを終了する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@classmethod
def close_session(
    cls, token: contextvars.Token[sqlalchemy.orm.Session], name: str | None = None, log_level: int = logging.DEBUG
) -> None:
    """セッションを終了する。"""
    session = cls.session()
    if name is not None:
        logger.log(
            log_level,
            f"セッション終了: {name} session={id(session):x},"
            f" thread={threading.get_ident():x},"
            f" task={pytilpack.asyncio.get_task_id_hex()}",
        )
    safe_close(session)
    cls.session_var.reset(token)
session() classmethod

セッションを取得する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
282
283
284
285
286
287
288
@classmethod
def session(cls) -> sqlalchemy.orm.Session:
    """セッションを取得する。"""
    sess = cls.session_var.get(None)
    if sess is None:
        raise RuntimeError(f"セッションが開始されていません。{cls.__qualname__}.start_session()を呼び出してください。")
    return sess
select() classmethod

sqlalchemy.Selectを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
290
291
292
293
294
@classmethod
def select(cls) -> sqlalchemy.Select[tuple[typing.Self]]:
    """sqlalchemy.Selectを返す。"""
    # cls.count()などでfrom句が消えないように明示的にfrom句を指定して返す。
    return sqlalchemy.select(cls).select_from(cls)
insert() classmethod

sqlalchemy.Insertを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
296
297
298
299
@classmethod
def insert(cls) -> sqlalchemy.Insert:
    """sqlalchemy.Insertを返す。"""
    return sqlalchemy.insert(cls)
update() classmethod

sqlalchemy.Updateを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
301
302
303
304
@classmethod
def update(cls) -> sqlalchemy.Update:
    """sqlalchemy.Updateを返す。"""
    return sqlalchemy.update(cls)
delete() classmethod

sqlalchemy.Deleteを返す。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
306
307
308
309
@classmethod
def delete(cls) -> sqlalchemy.Delete:
    """sqlalchemy.Deleteを返す。"""
    return sqlalchemy.delete(cls)
count(query) classmethod

queryのレコード数を取得する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
311
312
313
314
315
316
317
@classmethod
def count(cls, query: sqlalchemy.Select | sqlalchemy.CompoundSelect) -> int:
    """queryのレコード数を取得する。"""
    # pylint: disable=not-callable
    return (
        cls.scalar_one_or_none(sqlalchemy.select(sqlalchemy.func.count()).select_from(query.order_by(None).subquery())) or 0
    )
scalar_one(query) classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
@classmethod
def scalar_one[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).scalar_one()
scalar_one_or_none(query) classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
T | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
@classmethod
def scalar_one_or_none[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> T | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).scalar_one_or_none()
scalars(query) classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[tuple[T]] | CompoundSelect[tuple[T]]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[T]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
352
353
354
355
356
357
358
359
360
361
362
363
@classmethod
def scalars[T](cls, query: sqlalchemy.Select[tuple[T]] | sqlalchemy.CompoundSelect[tuple[T]]) -> list[T]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list(cls.session().execute(query).scalars().all())
one(query) classmethod

queryの結果を1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT]

1件のインスタンス。

発生：

タイプ デスクリプション
NoResultFound

結果が0件の場合。
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
@classmethod
def one[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT]:
    """queryの結果を1件取得する。

    Args:
        query: クエリ。

    Returns:
        1件のインスタンス。

    Raises:
        sqlalchemy.exc.NoResultFound: 結果が0件の場合。
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).one()
one_or_none(query) classmethod

queryの結果を0件または1件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
Row[TT] | None

0件の場合はNone、1件の場合はインスタンス。

発生：

タイプ デスクリプション
MultipleResultsFound

結果が複数件の場合。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
@classmethod
def one_or_none[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> sqlalchemy.Row[TT] | None:
    """queryの結果を0件または1件取得する。

    Args:
        query: クエリ。

    Returns:
        0件の場合はNone、1件の場合はインスタンス。

    Raises:
        sqlalchemy.exc.MultipleResultsFound: 結果が複数件の場合。

    """
    return cls.session().execute(query).one_or_none()
all(query) classmethod

queryの結果を全件取得する。

引数：

名前 タイプ デスクリプション デフォルト
query Select[TT] | CompoundSelect[TT]

クエリ。

 必須

戻り値：

タイプ デスクリプション
list[Row[TT]]

全件のインスタンスのリスト。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
398
399
400
401
402
403
404
405
406
407
408
409
@classmethod
def all[TT: tuple](cls, query: sqlalchemy.Select[TT] | sqlalchemy.CompoundSelect[TT]) -> list[sqlalchemy.Row[TT]]:
    """queryの結果を全件取得する。

    Args:
        query: クエリ。

    Returns:
        全件のインスタンスのリスト。

    """
    return list(cls.session().execute(query).all())
get_by_id_not_null(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。見つからない場合は例外を送出する。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用する。

 None

戻り値：

タイプ デスクリプション
Self

インスタンス。

Raises: ValueError: 見つからない場合。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
@classmethod
def get_by_id_not_null(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self:
    """IDを元にインスタンスを取得。見つからない場合は例外を送出する。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用する。

    Returns:
        インスタンス。
    Raises:
        ValueError: 見つからない場合。
    """
    instance = cls.get_by_id(id_, for_update=for_update, options=options)
    if instance is None:
        raise ValueError(f"{cls.__qualname__}が見つかりませんでした。id={id_}")
    return instance
get_by_id(id_, for_update=False, options=None) classmethod

IDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
id_ int

ID。

 必須
for_update bool

更新ロックを取得するか否か。

 False
options ExecutableOption | None

クエリオプション。eager loadingなどに使用。

 None

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
@classmethod
def get_by_id(
    cls, id_: int, for_update: bool = False, options: sqlalchemy.sql.base.ExecutableOption | None = None
) -> typing.Self | None:
    """IDを元にインスタンスを取得。

    Args:
        id_: ID。
        for_update: 更新ロックを取得するか否か。
        options: クエリオプション。eager loadingなどに使用。

    Returns:
        インスタンス。

    """
    q = cls.select().where(cls.id == id_)  # type: ignore  # pylint: disable=no-member
    if options is not None:
        q = q.options(options)
    if for_update:
        q = q.with_for_update()
    return cls.scalar_one_or_none(q)
paginate(query, page, per_page, scalar=True) classmethod

Flask-SQLAlchemy風ページネーション。

引数：

名前 タイプ デスクリプション デフォルト
query Select | CompoundSelect

ページネーションするクエリ。

 必須
page int

ページ番号。

 必須
per_page int

1ページあたりのアイテム数。

 必須
scalar bool

Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

 True

戻り値：

タイプ デスクリプション
Paginator

ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
@classmethod
def paginate(
    cls,
    query: sqlalchemy.Select | sqlalchemy.CompoundSelect,
    page: int,
    per_page: int,
    scalar: bool = True,
) -> pytilpack.paginator.Paginator:
    """Flask-SQLAlchemy風ページネーション。

    Args:
        query: ページネーションするクエリ。
        page: ページ番号。
        per_page: 1ページあたりのアイテム数。
        scalar: Trueの場合、スカラー値を返す。Falseの場合、全件のインスタンスを返す。

    Returns:
        ページネーションされた結果を返すpytilpack.paginator.Paginatorインスタンス。
    """
    if page <= 0:
        raise ValueError(f"ページ番号は1以上でなければなりません: page={page}")
    if per_page <= 0:
        raise ValueError(f"1ページあたりのアイテム数は1以上でなければなりません: per_page={per_page}")
    total = cls.count(query)
    page_query = query.offset((page - 1) * per_page).limit(per_page)
    items = cls.scalars(page_query) if scalar else cls.all(page_query)
    # pylint: disable=protected-access
    return pytilpack.paginator.Paginator(page=page, per_page=per_page, items=items, total=total)
commit() classmethod

セッションをコミットする。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
483
484
485
486
@classmethod
def commit(cls) -> None:
    """セッションをコミットする。"""
    cls.session().commit()

SyncUniqueIDMixin

self.unique_idを持つテーブルクラスにユニークID操作メソッドを付与するmixin。

generate_unique_id() classmethod

ユニークIDを生成する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
492
493
494
495
@classmethod
def generate_unique_id(cls) -> str:
    """ユニークIDを生成する。"""
    return secrets.token_urlsafe(32)
get_by_unique_id(unique_id, allow_id=False, for_update=False) classmethod

ユニークIDを元にインスタンスを取得。

引数：

名前 タイプ デスクリプション デフォルト
unique_id str | int

ユニークID。

 必須
allow_id bool

ユニークIDだけでなくID(int)も許可するかどうか。

 False
for_update bool

更新ロックを取得するか否か。

 False

戻り値：

タイプ デスクリプション
Self | None

インスタンス。
ソースコード位置： pytilpack/sqlalchemy/sync.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
@classmethod
def get_by_unique_id(
    cls: type[typing.Self],
    unique_id: str | int,
    allow_id: bool = False,
    for_update: bool = False,
) -> typing.Self | None:
    """ユニークIDを元にインスタンスを取得。

    Args:
        unique_id: ユニークID。
        allow_id: ユニークIDだけでなくID(int)も許可するかどうか。
        for_update: 更新ロックを取得するか否か。

    Returns:
        インスタンス。

    """
    assert issubclass(cls, SyncMixin)
    if allow_id and isinstance(unique_id, int):
        q = cls.select().where(cls.id == unique_id)  # type: ignore
    else:
        q = cls.select().where(cls.unique_id == unique_id)  # type: ignore
    if for_update:
        q = q.with_for_update()
    return cls.scalar_one_or_none(q)

wait_for_connection(url, timeout=180.0)

DBに接続可能になるまで待機する。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
def wait_for_connection(url: str, timeout: float = 180.0) -> None:
    """DBに接続可能になるまで待機する。"""
    state = _WaitForConnectionState(url, timeout)
    while True:
        try:
            engine = sqlalchemy.create_engine(url)
            try:
                with engine.connect() as connection:
                    result = connection.execute(sqlalchemy.text("SELECT 1"))
                    result.close()
            finally:
                engine.dispose()
            state.on_success()
            break
        except Exception as e:
            time.sleep(state.on_failure(e))

safe_close(session, log_level=logging.DEBUG)

例外を送出せずにセッションをクローズ。

ソースコード位置： pytilpack/sqlalchemy/sync.py 
543
544
545
546
547
548
549
550
551
552
def safe_close(
    session: sqlalchemy.orm.Session | sqlalchemy.orm.scoped_session,
    log_level: int | None = logging.DEBUG,
):
    """例外を送出せずにセッションをクローズ。"""
    try:
        session.close()
    except Exception:
        if log_level is not None:
            logger.log(log_level, "セッションクローズ失敗", exc_info=True)

run_sync_with_session(func)

同期関数を非同期に実行し、スレッド内でセッションを管理するデコレーター。

別スレッドで実行される関数内で自動的にセッションスコープを作成する。 各呼び出しは独立したセッション・トランザクションを持つ。

引数：

名前 タイプ デスクリプション デフォルト
func Callable[Concatenate[type[SyncMixin], P], R]

デコレート対象の同期関数

 必須

戻り値：

タイプ デスクリプション
Callable[Concatenate[type[SyncMixin], P], Awaitable[R]]

非同期版の関数
ソースコード位置： pytilpack/sqlalchemy/sync.py 
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
def run_sync_with_session[**P, R](
    func: "typing.Callable[typing.Concatenate[type[SyncMixin], P], R]",
) -> "typing.Callable[typing.Concatenate[type[SyncMixin], P], typing.Awaitable[R]]":
    """同期関数を非同期に実行し、スレッド内でセッションを管理するデコレーター。

    別スレッドで実行される関数内で自動的にセッションスコープを作成する。
    各呼び出しは独立したセッション・トランザクションを持つ。

    Args:
        func: デコレート対象の同期関数

    Returns:
        非同期版の関数
    """

    @functools.wraps(func)
    async def wrapper(cls: type[SyncMixin], *args: P.args, **kwargs: P.kwargs) -> R:
        def _impl() -> R:
            with cls.session_scope() as session:
                result = func(cls, *args, **kwargs)
                session.commit()
                return result

        return await asyncio.to_thread(_impl)

    return wrapper