|
| 1 | +from typing import Any, Optional |
| 2 | + |
1 | 3 | from sqlalchemy import Sequence
|
2 | 4 | from sqlalchemy.orm import Mapped, declarative_mixin, declared_attr, mapped_column
|
3 | 5 |
|
4 | 6 | from advanced_alchemy.types import BigIntIdentity
|
5 | 7 |
|
6 | 8 |
|
| 9 | +def _get_schema(cls: "BigIntPrimaryKey") -> Optional[str]: # pragma: nocover |
| 10 | + """Get the schema for the class if set via __table_args__, __table__, or __table_kwargs__.""" |
| 11 | + table_args = getattr(cls, "__table_args__", None) |
| 12 | + if isinstance(table_args, dict) and "schema" in table_args: |
| 13 | + return table_args["schema"] # type: ignore |
| 14 | + if isinstance(table_args, tuple) and table_args and isinstance(table_args[-1], dict) and "schema" in table_args[-1]: |
| 15 | + return table_args[-1]["schema"] # type: ignore |
| 16 | + if hasattr(cls, "__table__") and hasattr(cls.__table__, "schema"): # pyright: ignore |
| 17 | + return cls.__table__.schema # type: ignore[no-any-return] |
| 18 | + table_kwargs = getattr(cls, "__table_kwargs__", None) |
| 19 | + if isinstance(table_kwargs, dict) and "schema" in table_kwargs: |
| 20 | + return table_kwargs["schema"] # type: ignore |
| 21 | + return None |
| 22 | + |
| 23 | + |
7 | 24 | @declarative_mixin
|
8 | 25 | class BigIntPrimaryKey:
|
9 | 26 | """BigInt Primary Key Field Mixin."""
|
10 | 27 |
|
11 | 28 | @declared_attr
|
12 | 29 | def id(cls) -> Mapped[int]:
|
13 | 30 | """BigInt Primary key column."""
|
| 31 | + seq_kwargs: dict[str, Any] = {"optional": False} |
| 32 | + if schema := _get_schema(cls): |
| 33 | + seq_kwargs["schema"] = schema |
14 | 34 | return mapped_column(
|
15 | 35 | BigIntIdentity,
|
16 |
| - Sequence(f"{cls.__tablename__}_id_seq", optional=False), # type: ignore[attr-defined] |
| 36 | + Sequence(f"{cls.__tablename__}_id_seq", **seq_kwargs), # type: ignore[attr-defined] |
17 | 37 | primary_key=True,
|
18 | 38 | )
|
0 commit comments