sqlalchemy 相关问题

SQLAlchemy是一个Python SQL工具包和对象关系映射器,它为应用程序开发人员提供了SQL的全部功能和灵活性。

如何在 SQLAlchemy 中使用类。我一直在尝试使用类,但表没有创建

表未创建,我不想使用 declarative_base 和 sqlalchemy.orm 有人告诉我要上课,但我不知道如何安排。 类表(): def __init__(自身、引擎、元数据...

回答 1 投票 0

我如何在sqlalchemy中使用类,我一直在尝试使用类,但表没有创建

表未创建,我不想使用 declarative_base 和 alchemy.orm 有人告诉我要上课,但我不知道如何安排 类表(): def __init__(自身、引擎、元数据、...

回答 1 投票 0

更新实例后检查重复值

我有一个用户更新功能,我允许用户更改他们的电子邮件地址,但相同的地址在数据库中必须是唯一的,因此,在更新之前,我必须检查他们的新电子邮件是否已经存在...

回答 2 投票 0

如何检测sqlalchemy中自定义类型的变化

我正在使用 TypeDecorator 处理 sqlalchemy 列中的自定义类型。我将数据存储在 Postgres DB 内的 JSONB 中,但在代码中,我在数据类中对其进行序列化和反序列化。但当我

回答 1 投票 0

SqlAlchemy - 按关系属性过滤

我对 SQLAlchemy 没有太多经验,但我遇到了一个无法解决的问题。我尝试搜索并尝试了很多代码。 这是我的课程(简化为最重要的代码): 班级

回答 7 投票 0

Databricks SQLAlchemy TBLPROPERTIES

因此,我们使用 SQLAlchemy 创建了一个模型,并且希望将其部署在 DataBricks 上。我们模型中的表具有列的默认值。我们如何使用 SQLAlchemy 部署表并使用...

回答 1 投票 0

如何优化我的 SQLAlchemy 实例?

我正在使用 SQLAlchemy 来获取大量数据。无需赘述,我们有一个“交付”表,该表链接到其他几个表,然后这些表又链接到更多表。我...

回答 1 投票 0

运行时错误:当我调用 db.comit() 时事件循环已关闭

错误 失败的测试/users/api/test_auth.py::test_create_user_failed_unique_phone_number - RuntimeError:任务 错误 FAILED tests/users/api/test_auth.py::test_create_user_failed_unique_phone_number - RuntimeError: Task <Task pending name='Task-22' coro=<test_create_user_failed_unique_phone_number() running at /app/tests/users/api/test_auth.py:36> cb=[_run_until_complete_cb() at /usr/local/lib/python3.12/asyncio/base_events... 当我尝试调用 db.commit 时,总是发生错误。对于数据库连接,我使用数据库 url postgresql+asyncpg。 # conftest.py pytest_plugins = [ "tests.users.factories.users", ] async def init_db(engine): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async def close_db(engine): async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) @pytest.fixture(scope="session", autouse=True) async def db_engine(): engine = create_async_engine(settings.DATABASE_URL) await init_db(engine) yield engine await close_db(engine) await engine.dispose() @pytest.fixture async def db() -> AsyncIterable[AsyncSession]: async with SessionLocal() as session: yield session @pytest.fixture async def db_ins(db: AsyncSession) -> AsyncIterable[DBInsert]: async def insert(factory: SQLAlchemyModelFactory, **kwargs: Any) -> Any: instance = factory.build(**kwargs) db.add(instance) await db.commit() return instance yield insert @pytest.fixture(scope="session") async def client() -> AsyncIterable[AsyncClient]: async with AsyncClient(app=app, base_url="http://testserver") as client: yield client # tests.users.factories.users class UserFactory(SQLAlchemyModelFactory): class Meta: model = Users abstract = True first_name = factory.Faker("first_name") last_name = factory.Faker("last_name") email = factory.Faker("email") phone = factory.Sequence(lambda n: fake_phone_number()) is_active = True hash_password: str = factory.LazyAttribute(lambda _: PasswordHelper().hash("12345678")) class AdminUserFactory(UserFactory): role = Role.ADMIN @pytest.fixture async def admin_user(db_ins: DBInsert) -> Users: user = await db_ins(AdminUserFactory) return user 当我使用 admin_user 夹具时,我收到 RuntimeError: Event Loop is close 异常。当我们在 db_ins 夹具中调用 db.commit() 时,就会发生这种情况。 我想用数据库解决这个问题 这是最新版本的 pytest-asyncio (https://github.com/pytest-dev/pytest-asyncio/issues/706) 的问题。 当您使用不同范围的灯具时会发生这种情况。 要使其正常工作,请使用 pytest-asyncio 版本 0.21.1 或更低版本

回答 1 投票 0

在FastAPI sqlachemy中连接2个表返回查询对象/

我正在从 YouTube 学习 FastAPI,当我们使用 sqlalchemy 连接表时,我没有得到与视频相同的结果。该视频是 2 年前的,我试图找到解决方案......

回答 1 投票 0

DataFrame.to_sql 需要指定分区列,因为目标表已分区

你知道如何将 python 数据帧数据插入到分区 SQL 中吗? 此代码适用于非分区表。 df.to_sql('my_table', engine, if_exists='append', method='multi')

回答 1 投票 0

sqlalchemy.exc.OperationalError:(MySQLdb._exceptions.OperationalError)(3780,引用列...)

我的项目是一个基于Python、MySql和SQLAlchemy的项目。但是当我尝试执行该项目时,它给了我这个问题: sqlalchemy.exc.OperationalError: (MySQLdb._exceptions.OperationalError...

回答 1 投票 0

从表单提交将数据写入mySQL数据库

我正在尝试将数据写入到 mySQL 的表单提交中,但它不会出现在我的数据库中。 我正在学习编程,所以如果我没有提供足够的信息,请原谅我。 该...

回答 1 投票 0

SQLAlchemy:事务管理

这是示例代码 引擎 = create_engine(url="postgresql+psycopg2://postgres:0000@localhost:5432/db") SessionLocal = sessionmaker(自动提交= False,autoflush = False,bind =引擎) ...

回答 1 投票 0

使用 SQLAlchemy join 返回单个结果集

在我的 Flask 应用程序中,我想创建一个联接查询并使用 SQLAlchemy 返回单个(平面)结果集。 简单来说,我在 models.py 文件中创建了两个表: 用户类(db.Model): __桌子...

回答 1 投票 0

SQLAlchemy - 没有时区的时间

我试图将 SQLAlchemy 中表示为字符串的时间保存到 Postgresql 表中,该表的列类型为不带时区的时间。 我已将 SQLAlchemy 列定义为 Time(timezone=False)。

回答 1 投票 0

在 Flask Marshmallow 中过滤嵌套字段

我想在 Marshmallow 3 中过滤 is_active 列为 True 的嵌套字段 考虑以下场景 我有3张桌子 用户(ID、姓名) 组织(ID、名称) 组织用户(id,

回答 2 投票 0

SQLAlchemy 条件关系

我希望在两个 ORM 对象之间建立一对多关系,并使用在应用约束时链接到相同“多”对象的第二个关系来扩展它。 下面的例子可能

回答 1 投票 0

在 SQLAlchemy 中按空值过滤 postgres JSON 列

所以我有一个这样的模型: 类 myModel(基础): id = 列(整数,primary_key=True) 边框=列(JSONB) 如何查询没有边框的行?我试过了: 过滤器(

回答 6 投票 0

在 PostgreSQL 表中快速搜索字符串,输入略有错误

假设我在 PostgreSQL 中有一个包含列的用户表 名字 (PK) 姓氏(PK) 电子邮件 现在已经有数百万用户。 1个用户有该记录 (约翰·史密斯,[email protected]) 现在我...

回答 1 投票 0

sqlalchemy 在多个列中是唯一的

假设我有一个代表位置的类。地点“属于”客户。位置由 unicode 10 字符代码标识。 “位置代码”在位置中应该是唯一的...

回答 4 投票 0

© www.soinside.com 2019 - 2024. All rights reserved.