标签: flask-sqlalchemy

我正在开发一个类似的应用程序

facebook/
         __init__.py
         feed/
             __init__.py
             business.py
             views.py
             models/
                    persistence.py
                    user.py
         chat/
             __init__.py
             models.py
             business.py
             views.py
         config/
                dev.py
                test.py
                prod.py 

我希望有三个环境Dev,Test并且Production.
我有以下要求:
A)当我启动服务器.python runserver.py,我想提一提我想要的环境,连接- Dev,Test或Production.
b.)Dev&Production应该建立架构,只需要连接到机器
c.)我也想让我的测试连接到sqlite db,并创建架构,运行测试

如何以配置方式实现此目的,以便我不必硬编码与数据库相关的任何内容.

烧瓶中是否有任何好的图案？

目前我runerver.py的环境硬编码我不喜欢,

app = Flask(__name__)
app.config['SECRET_KEY'] = dev.SECRET_KEY

我正在寻找比我更好的想法

我正在测试我的应用时,我在使用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-Migrate,具有以下模型:

listpull/models.py

from datetime import datetime

from listpull import db


class Job(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    list_type_id = db.Column(db.Integer, db.ForeignKey('listtype.id'),
                             nullable=False)
    list_type = db.relationship('ListType',
                                backref=db.backref('jobs', lazy='dynamic'))
    record_count = db.Column(db.Integer, nullable=False)
    status = db.Column(db.Integer, nullable=False)
    sf_job_id = db.Column(db.Integer, nullable=False)
    created_at = db.Column(db.DateTime, nullable=False)
    compressed_csv = db.Column(db.LargeBinary)

    def __init__(self, list_type, created_at=None):
        self.list_type = list_type
        if created_at is None:
            created_at = datetime.utcnow()
        self.created_at = created_at

    def __repr__(self):
        return '<Job {}>'.format(self.id)


class ListType(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), unique=True, nullable=False)

    def __init__(self, …

我正在使用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) …

Factory Boy弃用了set_creation_function(参见ChangeLog 2.6.1)并建议开发人员

使用SomeFactory的_create()方法的替换替换factory.set_creation_function(SomeFactory,creation_function)

我有i)一些衍生工厂类和ii)我的数据库会话在另一个模块中实例化,所以我尝试用下面的第二个代码块替换https://github.com/mattupstate/overholt中的工作示例.PyCharm警告我"db"导入没有被使用,所以我怀疑在设置sqlalchemy_session时可能没有正确解除引用？

使用nosetests 1.3.7(但现在不推荐使用FactoryBoy的set_creation_function):

from myapp.core import db

def create_sqlalchemy_model_function(class_to_create, *args, **kwargs):
    entity = class_to_create(**kwargs)
    db.session.add(entity)
    db.session.commit()
    return entity

Factory.set_creation_function(create_sqlalchemy_model_function)

不使用nosetests 2.x(看起来db没有被正确引用？)

from factory.alchemy import SQLAlchemyModelFactory as Factory
from myapp.core import db

class Factory():
    class Meta:
        sqlalchemy_session = db.session

    def _create(cls, model_class, *args, **kwargs):
        entity = model_class(*args, **kwargs)
        db.session.add(entity)
        db.session.commit()
        return entity

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 …

我定义了一个 SqlAlchemy 模型

from sqlalchemy.dialects.postgresql import JSONB

class User(db.Model):
    __tablename__ = "user"
    id = db.Column(db.Integer, primary_key=True)
    nickname = db.Column(db.String(255), nullable=False)
    city = db.Column(db.String(255))
    contact_list = db.Column(JSONB)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

def add_user():
    user = User(nickname="Mike")
    user.contact_list = [{"name": "Sam", "phone": ["123456", "654321"]}, 
                         {"name": "John", "phone": ["159753"]},
                         {"name": "Joe", "phone": ["147889", "98741"]}]
    db.session.add(user)
    db.session.commit()

if __name__ == "__main__":
    add_user()

如何使用电话从我的contact_list 中检索姓名？例如，我有147889，我如何检索Joe？