标签: sqlmodel

我正在尝试通过 SQLModel 定义 JSON 列：

from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, JSON


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None
    meta: JSON

代码来自SQLModel，但通过“meta”属性进行扩展。

将上面的代码与示例代码的其余部分（设置 sqlite、添加数据）一起使用，出现以下错误：

RuntimeError: no validator found for <class 'sqlalchemy.sql.sqltypes.JSON'>, see `arbitrary_types_allowed` in Config

我尝试通过以下方式扩展代码

class Hero(SQLModel, table=True):
    [...]
    meta: JSON

    @validator('meta')
    def validate_json(v):
        return v

    class Config:
        arbitrary_types_allowed = True 

但这会导致另一个错误：

sqlalchemy.exc.CompileError: (in table 'hero', column 'meta'): Can't generate DDL for …

使用SQLModel如何让 alembic 识别以下模型？

from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

我一直在研究的一种方法是导入 Alembic 的 SQLalchemy 模型，但查看源代码我找不到如何做到这一点。

如何使 Alembic 与 SQLModel 模型一起使用？

我正在尝试获取一对多关系中匹配的行数。当我尝试时parent.children_count我得到：

sqlalchemy.exc.MissingGreenlet：greenlet_spawn尚未被调用；不能在这里调用await_only()。是否在意想不到的地方尝试了 IO？（此错误的背景位于：https ://sqlalche.me/e/14/xd2s ）

我添加了expire_on_commit=False但仍然遇到相同的错误。我怎样才能解决这个问题？

import asyncio
from uuid import UUID, uuid4
from sqlmodel import SQLModel, Relationship, Field
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

class Parent(SQLModel, table=True):
    id: UUID = Field(default_factory=uuid4, primary_key=True)
    children: list["Child"] = Relationship(back_populates="parent")
    @property
    def children_count(self):
        return len(self.children)

class Child(SQLModel, table=True):
    id: UUID = Field(default_factory=uuid4, primary_key=True)
    parent_id: UUID = Field(default=None, foreign_key=Parent.id)
    parent: "Parent" = Relationship(back_populates="children")

async def main():
    engine = create_async_engine("sqlite+aiosqlite://")
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    async with AsyncSession(engine) as …

我正在使用 SQLModel 库做一个简单的事情，select()就像他们的官方网站上描述的那样。但是我收到Column expression or FROM clause expected错误消息

from typing import Optional

from sqlmodel import Field, Session, SQLModel, create_engine, select

from models import Hero
    
sqrl = f"mysql+pymysql:///roo@asdf:localhost:3306/datab"

engine = create_engine(sqrl, echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def select_heroes():
    with Session(engine) as session:
        statement = select(Hero)
        results = session.exec(statement)
        for hero in results:
            print(hero)


def main():
    select_heroes()


if __name__ == "__main__":
    main()

这是我的models/Hero.py代码：

from datetime import datetime, date, time
from typing import Optional
from sqlmodel import Field, …

我正在尝试 SQLModel ( https://sqlmodel.tiangolo.com/ )，我发现我必须在多个字段之间创建复合索引，但我不知道如何使用 SQLModel 库来做到这一点。

数据库模型

我发现的唯一解决方法是直接使用 sqlalchemy Index，而不是 index=true （来自为唯一字段创建索引时的 SQLModel 文档 - ）

class Jump(SQLModel, table=True):
    """
    SQL Table abstraction: Jump
    Contains data belonging to a connection between a questionnaire-version
    and another questionnaire-version
    """

    origin_name: str = Field(primary_key=True)
    origin_version: int = Field()
    destination_name: str = Field()

    __table_args__ = (
        Index(
            "compound_index_origin_name_version_destination_name",
            "origin_name",
            "origin_version",
            "destination_name",
        ),
    )

当使用 FastAPI 和 SQLModel 进行模块化导入时，如果打开 /docs，我会收到以下错误：

类型错误：issubclass() arg 1 必须是一个类

• Python 3.10.6
• pydantic 1.10.2
• 快速API 0.85.2
• sqlmodel 0.0.8
• macOS 12.6

这是一个可重现的示例。

用户.py

from typing import List, TYPE_CHECKING, Optional
from sqlmodel import SQLModel, Field

if TYPE_CHECKING:
    from item import Item

class User(SQLModel):
    id: int = Field(default=None, primary_key=True)
    age: Optional[int]
    bought_items: List["Item"] = []

项目.py

from sqlmodel import SQLModel, Field

class Item(SQLModel):
    id: int = Field(default=None, primary_key=True)
    price: float
    name: str

主要.py

from fastapi import FastAPI

from user import User

app = …

在 FastAPI 应用程序中创建模型时，我正在努力获取创建 UUID 字段的语法。我正在使用SQLModel。

基本上，我的 models.py 文件如下所示：

from datetime import datetime
from typing import Optional
import uuid

from sqlalchemy import Column, DateTime
from sqlalchemy.dialects import postgresql as psql
from sqlmodel import SQLModel, Field


class ModelBase(SQLModel):
    """
    Base class for database models.
    """
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(sa_column=Column(DateTime(timezone=True), default=datetime.utcnow))
    updated_at: datetime = Field(sa_column=Column(DateTime(timezone=True),
                                 onupdate=datetime.utcnow, default=datetime.utcnow))


class UUIDModelBase(ModelBase, table=True):
    """
    Base class for UUID-based models.
    """
    uuid: uuid.UUID = Field(sa_column=Column(psql.UUID(as_uuid=True)), default=uuid.uuid4)

上述错误与

AttributeError: 'FieldInfo' object …

我试图使用fastapi users包快速向使用 PostgreSQL 数据库的 FastAPI 项目添加注册和身份验证系统。我用来asyncio创建异步函数。

一开始，我只使用了 sqlAlchemy，并且我在这里尝试了他们的示例。我将这些代码行添加到我的 app/app.py 中，以在服务器启动时创建数据库。一切都很顺利。表 users 是在我的数据库上创建的。

@app.on_event("startup")
async def on_startup():
    await create_db_and_tables()

由于我使用的是 SQLModel，因此我将FastAPI Users - SQLModel 的数据库适配器添加到我的虚拟 en 包中。我添加了这些行以便fastapi_users/db/__init__.py能够使用 SQL 模型数据库。

try:
    from fastapi_users_db_sqlmodel import (  # noqa: F401
        SQLModelBaseOAuthAccount,
        SQLModelBaseUserDB,
        SQLModelUserDatabase,
    )
except ImportError:  # pragma: no cover
    pass

我还修改了app/users.py, 以SQLModelUserDatabase代替 sqlAchemy 之一。

async def get_user_manager(user_db: SQLModelUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)

以及app/dp.py使用SQLModelUserDatabase，，SQLModelBaseUserDB这里是完整的代码 …

我正在尝试找到一种方法让 SqlModel 和 Alembic 协同工作。\n我的目标是不必手动编辑自动生成的 Alembic 迁移。

这是我的模型类：

class SongBase(SQLModel):\n    name: str\n    artist: str\n    label: str = Field(index=False)\n    year: Optional[int] = None\n\n\nclass Song(SongBase, table=True):\n    id: int = Field(default=None, primary_key=True, nullable=False)\n\n

创建初始迁移等后，我添加我的枚举。在深入挖掘https://github.com/tiangolo/sqlmodel/issues?q=is%3Aissue+is%3Aopen+enum后，这是我最接近的有效方法：

class SongType(enum.Enum):\n    Rock = "Rock"\n    Ballad = "Ballad"\n\n\nclass SongBase(SQLModel):\n    name: str\n    artist: str\n    label: str = Field(index=False)\n    song_type: SongType = Field(\n        sa_column=Column(\n            Enum(SongType),\n            default=None,\n            nullable=True,\n            index=False\n        )\n    )\n    year: Optional[int] = None\n\n\nclass Song(SongBase, table=True):\n    id: int = Field(default=None, primary_key=True, nullable=False)\n\n

添加song_type …

通过SQLAlchemy，可以使用UniqueConstraint声明在两个或多个列上创建约束。

使用SQLModel执行此操作的最惯用方法是什么？

例如：

from sqlmodel import SQLModel, Field


class SubAccount(SQLModel):
    id: int = Field(primary_key=True)
    name: str
    description: str
    user_id: int = Field(foreign_key="user.id")

我怎样才能强制它name对于给定的应该是唯一的user_id？