问题[sqlalchemy](coding)

是否有可能在一对多层次结构中获取一个/所有深嵌套子表？

假设一个用户已经登录。

u = db.session.execute(db.select(U).where(U.email == '[email protected]')).scalar()

我想查询X3用户中的一个/所有表。该怎么做？

                    <- X3 ->
    <- X1 -><- X2 ->
                    <- X3 ->
-- U1 ->
                        <- X3 ->
        <- X1 -><- X2 ->
                        <- X3 ->

--------------------------------

                        <- X3 ->
        <- X1 -><- X2 ->
                        <- X3 ->
-- U2 ->
                        <- X3 ->
        <- X1 -><- X2 ->
                        <- X3 ->

class U(UserMixin, db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(64), index=True, unique=True)
    # ...
    x1: Mapped[list["X1"]] = db.relationship(back_populates="u", cascade='all, delete')

class X1(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    # ...
    u: Mapped["U"] = db.relationship(back_populates="x1", cascade='all, delete')
    u_id: Mapped[int] = mapped_column(ForeignKey("u.id"))
    x2: Mapped[list["X2"]] = db.relationship(back_populates="x1", cascade='all, delete')

class X2(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    # ...
    x1: Mapped["X1"] = db.relationship(back_populates="x2", cascade='all, delete')
    x1_id: Mapped[int] = mapped_column(ForeignKey("x1.id"))
    x3: Mapped[list["X3"]] = db.relationship(back_populates="x2", cascade='all, delete')
    
class X3(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    # ...
    x2: Mapped["X2"] = db.relationship(back_populates="x3", cascade='all, delete')
    x2_id: Mapped[int] = mapped_column(ForeignKey("x2.id"))

谢谢。

我有一个名为 Candidate 的模型，我想返回它，其中两个字段引用另一个模型 StaffMember。这些最初来自我的数据库/ORM，作为 sqlalchemy 响应。

class StaffMember(BaseModel):
    fname: str
    sname: str

    @computed_field
    @property
    def full_name(self) -> str:
        return f"{self.fname} {self.sname}"

    model_config = ConfigDict(from_attributes=True)

class CandidateResponse(BaseModel):
    id: int
    fname: str
    sname: str
    dob: date
    created_on: datetime
    created_by: StaffMember
    updated_on: datetime
    updated_by: StaffMember

    model_config = ConfigDict(from_attributes=True)

我想返回一个 CandidateResponse dict/JSON 作为单级。通常，响应如下所示：

{
  "id": 2,
  "fname": "Joe01",
  "sname": "Bloggs",
  "dob": "2004-12-31",
  "created_on": "2025-02-17T09:34:27",
  "created_by": {
    "fname": "System",
    "sname": "Default",
    "full_name": "System Default"
  },
  "updated_on": "2025-02-17T09:34:27",
  "updated_by": {
    "fname": "System",
    "sname": "Default",
    "full_name": "System Default"
  }
}

但是，我不想要任何嵌套字段。有办法吗？——理想情况下，无需再次定义所有字段。

我正在寻找更像这样的回应：

{
  "id": 2,
  "fname": "Joe01",
  "sname": "Bloggs",
  "dob": "2004-12-31",
  "created_on": "2025-02-17T09:34:27",
  "created_by_fname": "System",
  "created_by_sname": "Default",
  "created_by_full_name": "System Default",
  "updated_on": "2025-02-17T09:34:27",
  "updated_by_fname": "System",
  "updated_by_sname": "Default",
  "updated_by_full_name": "System Default"
}

我可以使用 field_serializer，但我觉得这是一个混乱的解决方案，因为我需要创建多个计算字段（每个嵌套模型）。

我还想问：

1. 这是合适的做法吗？ - API 使用者期望一个扁平的对象（不带嵌套）
2. 我是否应该考虑其他方法？ - 我已经专注于此，但也许我正在考虑修复/改变错误的事情。

我搜索了整个互联网和一些人工智能机器人，寻找一个在我看来极其基本的功能，但无济于事。用于检查服务器数据库版本的 SQLAlchemy 命令是什么？

我知道我可以执行原始 SQL 查询，但确切的语法因 SQL 变体而异。以下是两个示例：

1. Postgres：SELECT version()
2. SQLite的：SELECT get_version()

期望响应：

PostgreSQL 15.7 on x86_64-pc-linux-gnu, compiled by gcc (GCC) 7.3.1 20180712 (Red Hat 7.3.1-12), 64-bit
...or...
PostgreSQL 15.7

我搜索了 SQLAlchemy 1.4 和 2.0 文档，但找不到它。肯定有办法吧？

我对 SQLModel 还很陌生，所以请多包涵。

我按照他们网站上的 sqlmodel 教程并创建了我自己的 SQLModel 子类：

class Vehicle(SQLModel, table=True):
   __tablename__ = "vehicle"

   id: int = Field(primary_key=True)
   url: str =  Field(unique=True)
   vehicle_name: str | None = Field(default=None, nullable=True)

然后我创建引擎并将该模型应用到数据库：

url = "< my database url >"
engine = sqlmodel.create_engine(url)
Vehicle.metadata.create_all(engine)

第一次运行此代码时，一切正常。表格已创建，所有定义的列都已存在。但如果我更改车辆模型上的某些属性并再次运行此代码，则表格上没有任何变化。

为什么会发生这种情况？我该如何解决？如果我对车辆模型进行了某些更改，我希望该更改能够被检测到并自动迁移到数据库。

我正在使用 Flask-SQLAlchemy（版本 2.5.1）和 Python 3.11 连接到 PostgreSQL 15 数据库。

如果表存在，我想使用 DROP TABLE IF EXISTS ... CASCADE 删除该表。这是我的代码：

from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError

try:
    sql = text("DROP TABLE IF EXISTS {table_name} CASCADE")
    db.session.execute(sql)
    db.session.commit()

except Exception as e:
    logger.error(f"Failed to drop {table_name} table: {str(e)}")

except SQLAlchemyError as e:
    db.session.rollback()
    logger.error(f"An error occurred: {str(e)}")

但是，当表不存在时，我没有看到 Python 中引发任何异常，即使 PostgreSQL 日志显示以下内容：

LOG:  statement: DROP TABLE IF EXISTS devices CASCADE
NOTICE:  table "devices" does not exist, skipping
LOG:  statement: COMMIT

由于该表不存在，PostgreSQL 会记录一个通知，但我的 Python 代码中不会传播任何异常。

我的问题是：

1. 当表不存在时，传递“原始”SQL 语句是否会引发异常？
2. 有没有办法将数据库的状态消息绑定到我的 Python 代码，以便我可以更明确地处理这种情况？
3. execute()我是否应该在和commit()命令之后手动检查数据库的错误代码？

我已经研究了一段时间，但没有任何收获。如果我从命令行运行“alembic upgrade head”，效果很好。我想在 FastAPI 中执行此操作，这样当它启动时，它会升级数据库架构。所以经过大量搜索，我想出了这个方法：

async def run_migrations():
  alembic_cfg = Config("alembic.ini")
  url = str(engine.url)
  async with engine.begin() as conn:
    await conn.run_sync(alembic_cfg.set_main_option, "sqlalchemy.url", url)
    await conn.run_sync(command.upgrade, alembic_cfg, "head")

我在 FastAPI 生命周期启动期间调用它。它给出了这个错误：

TypeError: Config.set_main_option() takes 3 positional arguments but 4 were given

我也尝试过：

await conn.run_sync(alembic_cfg.set_main_option("sqlalchemy.url", url))

但这让我：

TypeError: 'NoneType' object is not callable

run_sync 不喜欢我这样做吗？我没有向 set_main_option 传递 4 个参数。url 的值是正确的，因为我将其打印出来以确保无误。

或者有没有更简单/更好的方法从 fastapi 内部升级到 head？

谢谢

• SQLAlchemy 2.0.32
• uvicorn 0.30.6
• fastapi 0.112.1
• alembic 1.13.2

我已经写了这个端点：

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status
from starlette import status

# Local Imports
from database import SessionLocal
# from .auth import CreateUserRequest, db_dependency, user_dependency
from models import Patients, Users


router =APIRouter(
    prefix='/patient',
    tags=['patient']
)

async def get_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    db = SessionLocal()
    try:
        yield db
    finally:
        await db.close()


# Annotation for our database for dependency injection
db_dependency = Annotated[AsyncSession, Depends(get_db)]

async def get_curent_user(token: Annotated[str, Depends(oauth2_bearer)]):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get('sub')
        user_id: int = payload.get('id')
        role_id: int = payload.get('role_id')
        if email is None or user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate user."
            )
        return {'username': email, 'id': user_id, 'role_id': role_id}
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
        detail='Could not validate user.')

user_dependency = Annotated[dict, Depends(get_curent_user)]


class CreateUserRequest(BaseModel):
    username: str # Email is username
    password: str
    phone: str
    dob: date
    role_id: int


class CreatePatinetReuqest(BaseModel):
    name: str
    registration_date: date
    user_id: int


@router.post("/add_patient/")
async def add_patient(db: db_dependency, create_user_request: CreateUserRequest):
    # Create a new user (patient)
    create_user_model = Users(
        email=create_user_request.username,
        password=None,
        phone=create_user_request.phone,
        date_of_birth=create_user_request.dob,
        role_id=0,
    )
    try:
        db.add(create_user_model)
        await db.commit()
    except IntegrityError as er:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"User {create_user_request.username} already exists. Choose a different username."
        )
    await db.refresh(create_user_model)

    # Create a corresponding patient entry
    add_patient_model = Patients(
        registration_date=date.today(),
        user_id=create_user_model.id
    )

    try:
        db.add(add_patient_model)
        await db.commit()
    except IntegrityError as er:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Patient with {add_patient_model.patient_id} already exists."
        )
    await db.refresh(add_patient_model)

    return {
        "message": "Patient added successfully",
        "patient_id": add_patient_model.patient_id
    }

然后我有 main.py

from fastapi import FastAPI

from routers import patient


app = FastAPI()
app.include_router(patient.router)

但是我收到了无法处理实体错误 422。

我将密码设置为 None 因为在 Users 表中它是可空的。 loc 是什么？

我正在尝试在 Async Sqlalchemy 2.0 中构建动态查询。选项以列表形式显示，如下所示：

select_options=[selectinload(Table.entitity1).selectinload(Table2.entity2)]

然后我这样做：

query = select(model).filter(model.id == item_id).options(select_options[0])
result = await async_session.execute(query)

效果很好，我得到了正确的响应。但我希望能够继续将选项附加到该语句。我尝试了这个作为示例，并认为如果这有效，那么我可以循环遍历 select_options 并添加每个选项：

query = select(model)
query.filter(model.id == item_id)
query.options(select_options[0])

我没有收到任何错误，但响应忽略了选项。我如何才能继续添加 .options().options() 等？

谢谢

我需要使用sqlalchemy-citext扩展来处理我的模型，但我无法让它与 psycopg (v3) 一起工作。

应用程序在启动时抛出错误：

0.258 import psycopg2.extensions
0.258 ModuleNotFoundError: No module named 'psycopg2'

我已经安装了以下软件包：

[tool.poetry.dependencies]
...
psycopg = {extras = ["binary", "pool"], version = "3.1.13"}
sqlalchemy = {extras = ["asyncio"], version = "^2.0.31"}
sqlalchemy-citext = "^1.8.0"
...

据我所知，sqlalchemy-citext == 1.8.0它兼容psycopg >= 3

我也尝试过使用DOCKER。结果是一样的。

from python

RUN python -m pip install -U pip psycopg sqlalchemy-citext

RUN python -c "import citext"

我试图让 Hylang 和 SQLAlchemy 合作，但我似乎做不到。我认为这与 Flask-SQLAlchemy 如何使用元类解析数据库模型有关，但我不能 100% 确定。defcolumn我尝试创建所有s （定义如下）的字典，但这给了我同样的错误。这是我的代码：

(require hyrule [assoc defmacro/g!])
(import os)
(import subprocess)
(import sys)
(import flask [Flask redirect render_template request url_for])
(import flask-sqlalchemy [SQLAlchemy])
(import flask-migrate [Migrate])

(setv app (Flask __name__))
(setv path (os.path.join (os.path.expanduser "~/dbs") "development.db"))
(assoc app.config "SQLALCHEMY_DATABASE_URI" f"sqlite:///{path}")
(assoc app.config "SQLALCHEMY_TRACK_MODIFICATIONS" False)
(setv db (SQLAlchemy app))
(setv migrate (Migrate app db))

(defmacro defcolumn [name _column type #* args]
  (if args
      `(do {~name (db.Column ~type #** {~@args})})
      `{~name (db.Column ~type)}))

(defclass Stream [db.Model]
  (defcolumn "id" db.Column db.Integer "primary_key" True)
  (defcolumn "name" db.Column (db.String 240) "unique" True "nullable" False)
  (defcolumn "url" db.Column (db.String 240) "unique" True "nullable" False)
  (defcolumn "currently-playing" db.Column db.Boolean "unique" False "default" False)
  (defcolumn "created" db.Column
    (db.DateTime :timezone True)
    "unique" False
    "default" (db.func.now))

  (defcolumn "updated" db.Column
    (db.DateTime :timezone True)
    "unique" False
    "default" (db.func.now))

  (defn __init__ [self name url currently-playing]
    (setv self.name name)
    (setv self.url url)
    (setv self.currently-playing currently-playing))

  (defn __repr__ [self]
    f"<Stream {self.name}>"))

这是错误。正如您所看到的，SQLAlchemy 没有primary_key从我传入的 kwargs 中获取密钥：

danieljaouen on Daniels-MBP at …/pi-radio-hy via main (?)
 hy app.hy
Traceback (most recent call last):
  File "/opt/homebrew/bin/hy", line 8, in <module>
    sys.exit(hy_main())
             ^^^^^^^^^
  File "<frozen runpy>", line 286, in run_path
  File "<frozen runpy>", line 98, in _run_module_code
  File "<frozen runpy>", line 88, in _run_code
  File "/Users/danieljaouen/src/pi-radio-hy/app.hy", line 25, in <module>
    (defclass Stream [db.Model]
     ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/flask_sqlalchemy/model.py", line 92, in __init__
    super().__init__(name, bases, d, **kwargs)
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/flask_sqlalchemy/model.py", line 144, in __init__
    super().__init__(name, bases, d, **kwargs)
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/decl_api.py", line 196, in __init__
    _as_declarative(reg, cls, dict_)
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/decl_base.py", line 247, in _as_declarative
    return _MapperConfig.setup_mapping(registry, cls, dict_, None, {})
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/decl_base.py", line 328, in setup_mapping
    return _ClassScanMapperConfig(
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/decl_base.py", line 582, in __init__
    self._early_mapping(mapper_kw)
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/decl_base.py", line 369, in _early_mapping
    self.map(mapper_kw)
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/decl_base.py", line 1957, in map
    mapper_cls(self.cls, self.local_table, **self.mapper_args),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<string>", line 2, in __init__
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/util/deprecations.py", line 281, in warned
    return fn(*args, **kwargs)  # type: ignore[no-any-return]
           ^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 853, in __init__
    self._configure_pks()
  File "/opt/homebrew/Cellar/hy/0.28.0/libexec/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 1637, in _configure_pks
    raise sa_exc.ArgumentError(
sqlalchemy.exc.ArgumentError: Mapper Mapper[Stream(stream)] could not assemble any primary key columns for mapped table 'stream'

这是我第一次尝试将应用程序从 Python 翻译为 Hylang，因此非常感谢这里的任何帮助。