我有一个名为Person的ORM类,它包含一个person表:
在设置与db等的连接之后,我运行ff语句.
people = session.query(Person).all()
人员表不包含任何数据(至今),因此当我打印变量人员时,我得到一个空列表.
我将我的ORM类People中引用的表重命名为people_foo(不存在).
然后我再次运行脚本.我很惊讶在尝试访问不存在的表时没有抛出任何异常.
因此,我有以下两个问题:
如果它有帮助,我使用PostgreSQL作为数据库
[编辑]
我正在写一个包.在我的people脚本中,我有以下代码(在此缩写):
people = session.query(Person).all()
[EDIT2]
common模块正确设置记录器,我可以在导入common的其他模块中使用logger.
但是在dbfuncs模块中,我收到以下错误/警告:
没有找到记录器"sqlalchemy.engine.base.Engine"的处理程序
出于各种原因,我试图从数据库中获取对象集合,并将其传递给另一个未连接到数据库的进程.我的代码看起来像下面的那个,但我一直在
sqlalchemy.exc.UnboundExecutionError: Instance <MyClass at 0x8db7fec> is not bound to a Session; attribute refresh operation cannot proceed
当我尝试在get_list()方法之外查看列表中的元素时.
def get_list (obj):
sesson = Session()
lst = session.query(MyClass).all()
session.close()
return lst
但是,如果我使用它
def get_list_bis (obj)
session = Session()
return session.query(MyClass).all()
我能够使用元素,但担心会话的状态,因为它没有关闭.
我在这里错过了什么？
尝试使用to_sql将pandas数据帧写入MySQL表.以前一直在使用flavor ='mysql',但是它将来会被折旧并且想要开始转换到使用SQLAlchemy引擎.
示例代码:
import pandas as pd
import mysql.connector
from sqlalchemy import create_engine
engine = create_engine('mysql+mysqlconnector://[user]:[pass]@[host]:[port]/[schema]', echo=False)
cnx = engine.raw_connection()
data = pd.read_sql('SELECT * FROM sample_table', cnx)
data.to_sql(name='sample_table2', con=cnx, if_exists = 'append', index=False)
读取工作正常,但to_sql有错误:
DatabaseError:sql上的执行失败'SELECT name FROM sqlite_master WHERE type ='table'AND name =？;':字符串格式化过程中参数数量错误
为什么看起来它试图使用sqlite？sqlalchemy与mysql,特别是mysql.connector的正确使用是什么？
我也尝试将引擎作为连接传递,这给了我一个引用没有游标对象的错误.
data.to_sql(name='sample_table2', con=engine, if_exists = 'append', index=False)
>>AttributeError: 'Engine' object has no attribute 'cursor'
基本上是这个问题（尚未得到解答）的精炼版本。
我想声明的是，变量应该只采用TypedDict.
目前我正在定义一个单独的Literal类型来表示键，例如：
from typing import Literal, TypedDict
class MyTD(TypedDict):
a: int
b: int
mytd = MyTD(a=1, b=2)
key = "a"
mytd[key] # error: TypedDict key must be a string literal; expected one of ('a', 'b')
MyTDKeyT = Literal["a", "b"]
typed_key: MyTDKeyT = "b"
mytd[typed_key] # no error
我希望能够Literal出于想要最小化重复代码的所有常见原因替换定义。
伪代码：
key: Keys[MyTD] = "a"
mytd[key] # would be no error
not_key: Keys[MyTD] = "z" # error
有办法实现这一点吗？
为了澄清，鉴于 mypy 可以告诉我键类型需要是“a”或“b”的文字，我希望可能有一种不太容易出错的方法来将变量注释为该类型，而不是使用并排维护两个单独的键列表，一次在定义中TypedDict，一次在Literal …
我有以下项目结构
--some db:
--some db:
--alchemy:
-- __init__.py
--alembic:
-- versions
-- env.py
-- README.py
-- script.py
--migrations:
-- __init__.py
--models:
-- model_1
-- model_2
-- __init__.py
我尝试通过 alembic 自动生成迁移。
我Base在__init__.py模型中的文件夹
import sqlalchemy as sa
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base, declared_attr
metadata = sa.MetaData()
Base = declarative_base(metadata=metadata)
并导入这是 env.py
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from models import Base
config = context.config
fileConfig(config.config_file_name) …举个简单的例子：
from __future__ import annotations
import typing as t
class MyType:
def __init__(self, s: str, i: int) -> None:
self.s = s
self.i = i
class MyProto(t.Protocol):
s: str
i: int
class MyDict(t.TypedDict):
s: str
i: int
def my_serializer(inst: MyProto) -> MyDict:
return {"s": inst.s, "i": inst.i}
d = my_serializer(MyType("a", 1))
所有类型检查均通过。
现在我们可以说这MyType实际上是一个具有许多属性的 ORM 类，它是协议和字典类型的真实来源。每次将属性添加到类中时，都必须在 Protocol 类主体和 TypedDict 类主体中维护相同的注释，这感觉有点多余。
我想知道是否有一种方法可以集中定义类型注释并告诉 mypy 这些是协议和 dict 类的类型定义。
我试过这个：
class TypeMixin:
s: str
i: int
class MyProto(TypeMixin, t.Protocol):
pass
class MyDict(TypeMixin, t.TypedDict): …我有一个名为SQL Server的库存表中dbo.inventory包含Year，Month，Material和Stock_quantity。我每天都会收到一个新的库存计数作为 csv 文件，需要将其加载到dbo.inventory表格中。然而，我需要删除的记录在数据库中，如果Year与Month从csv文件已经在数据库中，以便退出，以避免加载多个盘点同月。
在 SQL 中，我会这样做：
Delete t1
FROM dbo.inventory t1
JOIN csv t2 ON t1.Year = t2.Year and t1.Month = t2.Month
我不知道如何做到这一点在Python脚本，所以我不会调入我的CSV文件作为一个临时表到数据仓库，只是删除现有的行匹配Year和Month再加载它们。
我在另一个设置中使用了以下内容：
delete_date = sales.Date.max()
connection = engine.connect()
connection.execute(f"""delete from sales where Date = '{delete_date}'""")
connection.close()
但这在这里不起作用，因为应该删除的输入是一个数据框，理论上它可以包含多个年份和月份，如果它是对早期加载的数字的更正。
我正在尝试使用alembic将'id'主键列添加到已存在的MySQL表中.我试过以下......
op.add_column('mytable', sa.Column('id', sa.Integer(), nullable=False))
op.alter_column('mytable', 'id', autoincrement=True, existing_type=sa.Integer(), existing_server_default=False, existing_nullable=False)
但得到以下错误
sqlalchemy.exc.OperationalError: (OperationalError) (1075, 'Incorrect table definition; there can be only one auto column and it must be defined as a key') 'ALTER TABLE mytable CHANGE id id INTEGER NOT NULL AUTO_INCREMENT' ()
看起来像是由alembic生成的sql语句没有PRIMARY KEY在alter语句的末尾添加.我可以错过一些设置吗？
提前致谢!
请进行健康检查！
我想包括不正确的函数调用返回的确切消息时，了解意外测试失败match的参数pytest.raises()。
文档状态：
match – 如果指定，则断言异常与文本或正则表达式匹配
下面的 repl 中的指令序列几乎说明了一切，但由于某种原因，最后一个测试失败了。
PS C:\Users\peter_000\OneDrive\git\test> pipenv run python
Loading .env environment variables…
Python 3.6.4 (v3.6.4:d48eceb, Dec 19 2017, 06:54:40) [MSC v.1900 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>>
>>>
>>>
>>> import pytest
>>> pytest.__version__
'4.4.1'
>>>
>>> with pytest.raises(TypeError, match='a string'):
... raise TypeError('a string') # passes
...
>>> def func():
... pass
...
>>> func(None)
Traceback (most recent call …我是第一次尝试一起使用 Flask 应用程序工厂模式和 pytest 框架。我开始对 sqlite db 后端进行基本的健全性测试，尽管到目前为止测试工作正常并且我看到测试 db 文件已成功创建，但 falsk_sqlalchemy 告诉我它没有定义 db 后端。我试图找到 pdb 和交互式控制台的问题 - 一切看起来都很正常。看起来它与任何人都可以帮助我了解问题出在哪里有关？
(venv) C:\Users\dv\PycharmProjects\ste-speach-booking>python -m pytest tests/
=========================== test session starts ============================
platform win32 -- Python 3.6.8, pytest-5.1.1, py-1.8.0, pluggy-0.12.0
rootdir: C:\Users\dv\PycharmProjects\ste-speach-booking
collected 3 items
tests\test_models.py ... [100%]
============================= warnings summary =============================
tests/test_models.py::test_init
C:\Users\d837758\PycharmProjects\ste-speach-booking\venv\lib\site-packages\flask_sqlalchemy\__init__.py:814: UserWarning: Neither SQLALCHEMY_DATABASE_URI nor SQLALCHEMY_BINDS is set. Defaulting SQLALCHEMY_DATABASE_URI to "sqlite:///:memory:".
'Neither SQLALCHEMY_DATABASE_URI nor SQLALCHEMY_BINDS is set. '
test_models 中的初始测试：
import pytest
import src.models
import datetime
def test_ActionTypes(db):
actiontype1 = …