标签: flask-sqlalchemy

我刚刚开始使用Flask/Flask-WTF/SQLAlchemy,我看到的大多数示例CRUD代码都显示了用于添加/编辑的单独模板.有两个模板几乎完全相同的形式html似乎重复(例如books_add.html,books_edit.html).从概念上讲,拥有一个模板(例如"books_form.html")对我来说更有意义,只需从两个单独的路径定义中调用同一模板上的render_template.我不太确定如何实现它,例如:

@app.route('/books/add')
def add_book():
...
render_template('books_form.html', action = 'add')


@app.route('/books/edit/<id>')
def edit_book(id):
...
render_template('books_form.html', action = 'edit', id = id)

但我不确定我是否走上了正确的轨道,或者偏离了最佳实践.任何输入都很受欢迎 - 关于如何处理单个模板文件以处理添加或编辑行为的具体想法.也欢迎链接到示例.

谢谢!

我正在使用Flask和SQLAlchemy构建一个应用程序.我基本上有3个表:用户,友谊和bestFriends:

用户可以拥有许多朋友,但只有一个最好的朋友.所以我希望我的模型是关系型的."一对多"用于"用户"和"友谊"之间的关系以及"用户"和"最佳朋友"之间的"一对一"关系.

这是我的模特:

from app import db
from sqlalchemy.orm import relationship, backref
from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

class users(db.Model):

    __tablename__ = "Users"

    id = db.Column(db.Integer, primary_key=True)
    userName = db.Column(db.String, nullable=False)
    userEmail = db.Column(db.String, nullable=False)
    userPhone = db.Column(db.String, nullable=False)
    userPass = db.Column(db.String, nullable=False)

    friendsR = db.relationship('friendships', backref='friendships.friend_id', primaryjoin='users.id==friendships.user_id', lazy='joined')

    def __init__(self, userName, userEmail, userPhone, userPass):
        self.userName = userName
        self.userEmail = userEmail
        self.userPhone = userPhone
        self.userPass = userPass

    def __repr__(self):
        return '{}-{}-{}-{}'.format(self.id, self.userName, self.userEmail, self.userPhone)

class …

我有一个像这样的鞋子模型:

class Shoe(db.Model):
id = db.Column(db.Integer, primary_key = True)
asin = db.Column(db.String(20), index = True)

我有一个像ids = [2,1,3]的id列表,当我在Shoe模型上查询结果在'ids'列表中有id时,我想回来:[{id:2,asin:" 111"},{id:1,asin:"113"},{id:3,asin:"42"}]但问题是使用以下查询语句不保留原始顺序,结果将会到来随机回来.如何保持我过滤的列表顺序？

不正确的一个: Shoe.query.filter(Shoe.id.in_(my_list_of_ids)).all()

在事件映射器级别文档中 它表示Session.add()不支持,但当我尝试在db.session.add(some_object)内部after_insert事件中工作时,例如:

def after_insert_listener(mapper, connection, user):
    global_group = Group.query.filter_by(groupname='global').first()
    a = Association(user,global_group)
    db.session.add(a)

event.listen(User, 'after_insert', after_insert_listener)

基本上任何新用户都应该是global_group的一部分,所以我在after_insert事件中添加了它.我试图插入一个用户,然后检查我的数据库,我找到了用户记录和关联记录.

我不知道这是否可以完成，但我正在尝试模拟我的 db.session.save。

我正在使用烧瓶和烧瓶炼金术。

数据库文件

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

单元测试

def test_post(self):
    with app.app_context():
        with app.test_client() as client:
            with mock.patch('models.db.session.save') as mock_save:
                with mock.patch('models.db.session.commit') as mock_commit:

                    data = self.gen_legend_data()
                    response = client.post('/legends', data=json.dumps([data]), headers=access_header)

                    assert response.status_code == 200
                    mock_save.assert_called()
                    mock_commit.assert_called_once()

和方法：

def post(cls):
    legends = schemas.Legends(many=True).load(request.get_json())

    for legend in legends:
        db.session.add(legend)

    db.session.commit()

    return {'message': 'legends saved'}, 200

我正在尝试模拟 db.session.add 和 db.session.commit。我试过db.session.save和legends.models.db.session.save和models.db.session.save。他们都带着保存错误回来了：

ModuleNotFoundError: No module named 'models.db.session'; 'models.db' is not a package

我没有收到错误，也不知道如何解决。 …

我正在测试我的应用时,我在使用Flask-Admin创建蓝图时遇到了麻烦.

这是我的View类(使用SQLAlchemy)

##
# All views that only admins are allowed to see should inherit from this class.
#
class AuthView(ModelView):
    def is_accessible(self):
        return current_user.is_admin()

class UserView(AuthView):
    column_list = ('name', 'email', 'role_code')

这是我初始化视图的方式:

# flask-admin
admin.add_view(UserView(User, db.session))
admin.init_app(app)

但是,当我尝试运行多个测试(故障总是发生在第二个测试和随后的所有其他测试)时,我总是收到以下错误消息:

======================================================================
ERROR: test_send_email (tests.test_views.TestUser)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/lib/python2.7/site-packages/nose/case.py", line 133, in run
    self.runTest(result)
  File "/lib/python2.7/site-packages/nose/case.py", line 151, in runTest
    test(result)
  File "/lib/python2.7/site-packages/flask_testing.py", line 72, in __call__
    self._pre_setup()
  File "/lib/python2.7/site-packages/flask_testing.py", line 80, in _pre_setup
    self.app = self.create_app()
  File …

我正在使用Flask-SQLAlchemy 1.0,Flask 0.10,SQLAlchemy 0.8.2和Python 2.7.5.我正在使用Oracle的MySQL Connector/Python 1.0.12连接MySQL 5.6.

当我重新启动我的Web服务器(Apache2或Flask的内置)时,我OperationalError: MySQL Connection not available在MySQL wait_timeout过期后收到异常(默认为8小时).

我发现有类似问题并明确设置的人SQLALCHEMY_POOL_RECYCLE = 7200,即使这是Flask-SQLAlchemy的默认设置.当我在这里放置断点时,我看到拆解功能session.remove()在每个请求后成功调用.有任何想法吗？

2014年7月21日更新:

由于这个问题继续受到关注,我必须补充一点,我确实尝试了一些提案.我的两次尝试看起来如下:

第一:

@contextmanager
def safe_commit():
    try:
        yield
        db.session.commit()
    except:
        db.session.rollback()
        raise

这允许我像这样包装我的提交调用:

with safe_commit():
    model = Model(prop=value)
    db.session.add(model)

我99%肯定我没有错过db.session.commit这种方法的任何电话,我仍然有问题.

第二:

def managed_session():
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            try:
                response = f(*args, **kwargs)
                db.session.commit()
                return response
            except:
                db.session.rollback()
                raise
            finally:
                db.session.close()
        return decorated_function
    return decorator …

我正在使用Flask-SQLAlchemy来做一个相当大的60k行的批量插入.我在这张桌子上也有多对多的关系,所以我不能用db.engine.execute它.在插入之前,我需要在数据库中找到类似的项目,如果找到重复的项目,则将插入更改为更新.

我可以事先做这个检查,然后通过插件进行批量插入db.engine.execute,但是在插入时我需要该行的主键.

目前,我做了db.session.add(),并db.session.commit()在每次插入,我也得到每秒进账3-4插入.

我运行了一个分析器,看看瓶颈在哪里,似乎db.session.commit()占用了60%的时间.

有没有什么方法可以让我更快地完成这个操作,可能是通过分组提交,但是哪个会给我主键？

这就是我的模型:

class Item(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(1024), nullable=True)
    created = db.Column(db.DateTime())
    tags_relationship = db.relationship('Tag', secondary=tags, backref=db.backref('items', lazy='dynamic'))
    tags = association_proxy('tags_relationship', 'text')

class Tag(db.Model):
id = db.Column(db.Integer, primary_key=True)
text = db.Column(db.String(255))

我的插入操作是这样运行的:

for item in items:
    if duplicate:
        update_existing_item
    else:
        x = Item()
        x.title = "string"
        x.created = datetime.datetime.utcnow()
        for tag in tags:
            if not tag_already_exists:
                y = Tag()
                y.text = "tagtext"
                x.tags_relationship.append(y) …

TL; DR:我如何让sqmbic理解并为sqlalchemy中创建的物化视图生成SQL？

我正在使用flask-sqlalchemy并使用alembic和postgres.为了获得使用sqlalchemy的物化视图,我在该主题上发表了一篇很好的帖子.我使用它很多,只有一些小的分歧(文章也使用flask-sqlalchemy,但完整的代码示例直接使用sqlalchemy的声明性基础).

class ActivityView(db.Model):
    __table__ = create_materialized_view(
        'activity_view',
        db.select([
            Activity.id.label('id'),
            Activity.name.label('name'),
            Activity.start_date.label('start_date'),
        ]).where(
            db.and_(
                Activity.start_date != None,
                Activity.start_date <=
                    datetime_to_str(datetime.now(tz=pytz.UTC) + timedelta(hours=48))
            )
        )
    )

    @classmethod
    def refresh(cls, concurrently=True):
        refresh_materialized_view(cls.__table__.fullname, concurrently)

db.Index('activity_view_index',
         ActivityView.__table__.c.id, ActivityView.__table__.c.start_date,
         unique=True)

这些create_materialized_view和refresh_materialized_view方法直接来自博客文章.

请注意,上面的示例已经大大简化,并且由于我的简化可能看起来很愚蠢,但我想要了解的真正想法是如何在迁移期间将alembic转换为一系列的alembic操作？

当我运行测试时,代码运行正常,视图生成正常,一切正常.当alembic运行时,它不会对视图执行任何操作.所以我最终要做的是将测试为物化视图发出的SQL复制到alembic迁移/版本文件中,然后最终直接执行:

op.execute(activities_view_sql)

类似地,我在物化视图上生成唯一索引时执行相同的直接SQL执行.

不幸的是,我的方法容易出错,并且会产生看似不必要的代码重复.

有没有办法让alembic了解我,ActivityView以便任何时候它改变,alembic将知道如何更新视图？

非常感谢!

假设您有以下简化示例模式,它使用SQLAlchemy加入表多态继承.Engineer和Analyst模特有Role关系.该Intern模型没有.

class Role(db.Model):

    __tablename__ = 'role'

    id = db.Column(db.Integer, primary_key=True)

    name = db.Column(db.String(16), index=True)

class EmployeeBase(db.Model):

    __tablename__ = 'employee_base'

    id = db.Column(db.Integer, primary_key=True)

    some_attr = db.Column(db.String(16))
    another_attr = db.Column(db.String(16))

    type = db.Column(db.String(50), index=True)

    __mapper_args__ = {
        'polymorphic_identity': 'employee',
        'polymorphic_on': type
    }

class Engineer(EmployeeBase):

    __tablename__ = 'engineer'

    id = db.Column(db.Integer, db.ForeignKey('employee_base.id'), primary_key=True)

    role_id = db.Column(db.Integer, db.ForeignKey('role.id'), index=True)
    role = db.relationship('Role', backref='engineers')

    __mapper_args__ = {
        'polymorphic_identity': 'engineer',
    }

class Analyst(EmployeeBase):

    __tablename__ = 'analyst'

    id …