标签: sqlalchemy

我在我的网络项目中使用SQlAlchemy.我应该使用scoped_session(session_maker())或plain session_maker()以及为什么？或者我应该使用其他东西？

## model.py
from sqlalchemy import *
from sqlalchemy.orm import *

engine = create_engine('mysql://dbUser:dbPassword@dbServer:dbPort/dbName',
pool_recycle=3600, echo=False)
metadata = MetaData(engine)
Session = scoped_session(sessionmaker())
Session.configure(bind=engine)
user = Table('user', metadata, autoload=True)

class User(object):
pass

usermapper = mapper(User, user)

## some other python file called abc.py
from models import *

def getalluser():
   session = Session()  
   session.query(User).all()
   session.flush()
   session.close()

## onemore file defg.py
from models import *

def updateuser():
   session = Session()  
   session.query(User).filter(User.user_id == '4').update({User.user_lname: 'villkoo'})
   session.commit()
   session.flush()
   session.close()

我为每个请求创建一个session = Session()对象,然后关闭它.我做对了还是有更好的方法吗？

我在我的python代码中收到了如下错误(它收集了twitter状态并存储在数据库中).

sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back by a nested rollback() call.  To begin a new transaction, issue Session.rollback() first.

我想知道问题是什么,它为什么会发生,我该如何解决它.

我不知道nested rollback.有没有简单的例子nested rollback？

编辑:问题解决了.在这种情况下,重点是在我们显式调用rollback之前不会执行回滚,因此当我们包含时nested rollback,我们应该在try语句中编写它,并nested rollback在http:// www中写入异常语句(在大多数情况下).sqlalchemy.org/TRAC /维基/ FAQ#Thetransactionisinactiveduetoarollbackinasubtransaction

而且,这是正确的代码示例.我从上面的链接中引用了这个.

sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back by a nested rollback() call.  To begin a new transaction, issue Session.rollback() first.

我正在使用Flask-Alchemy构建一个针对Cloud SQL的GAE Flask应用程序,并在dev_appserver构建它时运行以测试应用程序.

但是,如果我将URL 设置SQLALCHEMY_DATABASE_URI为mysql+gaerdbms:///appname?instance=instanceidURL,则在尝试调用时会收到以下回溯db.create_all():

Traceback (most recent call last):
  # earlier lines omitted for brevity
  File "/Project/app/foo.bar/foo/bar/admin/__init__.py", line 26, in init_db
    db.create_all()
  File "/Project/app/distlib/flask_sqlalchemy/__init__.py", line 856, in create_all
    self._execute_for_all_tables(app, bind, 'create_all')
  File "/Project/app/distlib/flask_sqlalchemy/__init__.py", line 848, in _execute_for_all_tables
    op(bind=self.get_engine(app, bind), tables=tables)
  File "/Project/app/distlib/flask_sqlalchemy/__init__.py", line 797, in get_engine
    return connector.get_engine()
  File "/Project/app/distlib/flask_sqlalchemy/__init__.py", line 473, in get_engine
    self._engine = rv = sqlalchemy.create_engine(info, **options)
  File "/Project/app/distlib/sqlalchemy/engine/__init__.py", line 332, in create_engine
    return strategy.create(*args, **kwargs)
  File "/Project/app/distlib/sqlalchemy/engine/strategies.py", …

我有一个使用 sqlalchemy 将数据保存到数据库的小型 Python Web 应用程序（用 Flask 编写）。当我尝试插入重复行时，会引发异常，如下所示：

(psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "uix_my_column"

我想包装异常并重新引发我自己的异常，以便我可以添加我自己的特定于该特定错误的日志记录和消息传递。这是我尝试过的（简化）：

from db import DbApi
from my_exceptions import BadRequest
from psycopg2.errors import UniqueViolation # <-- this does not exist!

class MyClass:
    def __init__(self):
        self.db = DbApi() 

    def create(self, data: dict) -> MyRecord:
        try:
            with self.db.session_local(expire_on_commit=False) as session:
                my_rec = MyRecord(**data)
                session.add(my_rec)
                session.commit()
                session.refresh(my_rec)
                return my_rec
        except UniqueViolation as e:
            raise BadRequest('A duplicate record already exists')

但这无法捕获错误，因为psycopg2.errors.UniqueViolation它实际上不是类名 (!)。

在 PHP 中，这就像捕获复制/粘贴异常的类名一样简单，但在 Python 中，这更加模糊。 …

我有一个关于Python和sqlalchemy模块的问题.cursor.rowcountsqlalchemy Python中的等价物是什么？

INSERT ... ON DUPLICATE KEY UPDATE在SQLAlchemy中有一种优雅的方式吗？我的意思是语法类似的东西inserter.insert().execute(list_of_dictionaries)？

我的印象是,通过SQLAlchemy进行的数据库调用将阻塞,并且不适用于除同步代码之外的任何其他内容.我是否正确(我希望我不是!)或者有没有办法将其配置为非阻塞？

我有两个Flask-SQLAlchemy模型,具有简单的一对多关系,如下面的最小示例:

class School(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30))
    address = db.Column(db.String(30))

class Teacher(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30))
    id_school = db.Column(db.Integer, db.ForeignKey(School.id))

    school = relationship('School', backref='teachers')

然后我向使用该关系的教师添加混合属性,如:

@hybrid_property
def school_name(self):
    return self.school.name

当我使用它时,该属性工作得很好teacher_instance.school_name.但是,我也想提出类似的查询Teacher.query.filter(Teacher.school_name == 'x'),但这给了我一个错误:

`AttributeError: Neither 'InstrumentedAttribute' object nor 
'Comparator' object has an attribute 'school_name'`. 

在SQLAlchemy文档之后,我添加了一个简单的混合表达式,如下所示:

@school_name.expression
def school_name(cls):
    return School.name

但是,当我再次尝试相同的查询时,它会生成一个没有join子句的SQL查询,因此我获得了School中所有可用的行,而不仅仅是那些与Teacher中的外键匹配的行.

从SQLAlchemy文档中我意识到表达式需要一个已经存在连接的上下文,所以我再次尝试查询:

Teacher.query.join(School).filter(Teacher.school_name == 'x')

这实际上是有效的,但是如果我需要学校模型的知识来实现​​这一目的,那就首先试图在那里试图获得语法糖.我希望有一种方法可以在表达式中加入,但我无法在任何地方找到它.该文档有一个示例,表达式返回一个直接用它构建的子查询select(),但即使这对我没用.

有任何想法吗？

UPDATE

在下面的Eevee回答之后,我使用了建议的关联代理并且它有效,但我也对它应该与select()子查询一起工作的评论感到好奇,并试图弄清楚我做错了什么.我最初的尝试是:

@school_name.expression
def school_name(cls):
    return select(School.name).where(cls.id_school == School.id).as_scalar() …

我正在运行基于查询的其他ID的查询.我遇到的问题是,有时查询将找不到结果.而不是让整个程序崩溃,我如何检查结果是否为None？

这是我的查询:

sub_report_id = DBSession.query(TSubReport.ixSubReport).filter(and_(TSubReport.ixSection==sectionID[0], TSubReport.ixReport== reportID[0])).one()

当代码执行并且没有找到结果时,我得到一个NoResultFound异常

NoResultFound: No row was found for one()

如果没有结果,有没有办法跳过查询？

在SO上找到解决方案(之前找不到) 从sqlalchemy获取第一行

我正在使用SqlAlchemy和Flask-migrate进行数据库迁移.我已经成功init了DB和upgrade一次,但是当我删除了我的表列的一个,我设法migrate但是upgrade给了我以下错误:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "DROP": syntax error [SQL: u'ALTER TABLE posts DROP COLUMN tags']

我的models.py有一部分

class Post(db.Model):
    __tabelname__ = 'posts'
    id = db.Column(db.Integer, primary_key=True)
    body = db.Column(db.UnicodeText)
    # tags = db.Column(db.Unicode(32))
    # I deleted this field, upgrade give me error

    ....

然后我再次运行python manage.py db upgrade,错误改变了!

(venv)ncp@ubuntu:~/manualscore$ python manage.py db upgrade
INFO  [alembic.migration] Context impl SQLiteImpl.
INFO  [alembic.migration] Will assume non-transactional DDL.
INFO  [alembic.migration] Running upgrade 555b78ffd5f -> 2e063b1b3164, add tag …