标签: flask-sqlalchemy

我有两个SQLAlchemy模型设置如下:

##############
# Post Model #
##############
class Post(db.Model):
    id = db.Column(db.Integer, primary_key = True)
    title = db.Column(db.String(250))
    content = db.Column(db.String(5000))
    timestamp = db.Column(db.Integer)
    author_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    likes = db.relationship('Like', backref = 'post', lazy = 'dynamic')

###############
# Likes Model #
###############
class Like(db.Model):
    id = db.Column(db.Integer, primary_key = True)
    voter_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    post_id = db.Column(db.Integer, db.ForeignKey('post.id'))

正如您所看到的,这些帖子上有帖子模型和用户喜欢的模型.我想创建一个选择所有帖子的查询,按照帖子的喜欢数量排序.在shell中,我可以运行:

SELECT post.*, 
       count(like.id) AS num_likes
FROM post
LEFT JOIN like
ON post.id = like.post_id
GROUP BY post.id;

什么是等效的SQLAlchemy命令？

谢谢!

这是我的Budget架构

class Budget(db.Model):
    __tablename__ = 'budgets'
    # noinspection PyRedeclaration
    uuid = Column('uuid', GUID(), default=uuid.uuid4, primary_key=True,
                  unique=True)
    user_id = Column(GUID(), ForeignKey('users.uuid'), nullable=False)
    user = relationship('User', backref='budgets')
    created_on = Column('created_on', sa.types.DateTime(timezone=True),
                        nullable=False)

和 BudgetCategories

class BudgetCategory(db.Model):
    __tablename__ = 'budget_categories'
    # noinspection PyRedeclaration
    uuid = Column('uuid', GUID(), default=uuid.uuid4, primary_key=True,
                  unique=True)
    budget_id = Column(GUID(), ForeignKey('budgets.uuid'), nullable=False)
    budget = relationship('Budget', backref='budgetCategories',
                          cascade="all, delete-orphan", single_parent=True)
    category = Column('category', sa.types.String, nullable=True)
    parent_category = Column('parent_category', sa.types.String, nullable=True)
    amount = Column('amount', Numeric(10, 2), nullable=False)
    recurring = Column('recurring', …

我正在使用Flask,Alembic和PostgreSQL与SQLAlchemy.我有location_messages一个列的现有表campaign_id.这是最初在模型中使用代码创建的

campaign_id = db.Column(db.Integer)

我想为它添加一个外键,所以我更新了模型

campaign_id = db.Column(db.Integer, db.ForeignKey('campaigns.id'))

我跑了revision --autogenerate但它没有创建任何东西 - 所以我一直在查看文档,但我不能理解我的用法语法.对于它的价值,在Alembic中从头开始创建表(放下它之后)是

op.create_table('location_messages',
[...]
sa.Column('campaign_id', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['campaign_id'], ['campaigns.id'], ),
sa.PrimaryKeyConstraint('id')
)

但对于我的生活,我找不到一种方法来为现有的桌子做这件事.有没有办法获取表的实例并将其分配ForeignKeyConstraint给列？

我正在尝试使用Flask-SQLAlchemy构建Flask应用程序; 我使用pytest来测试数据库.其中一个问题似乎是在不同测试之间创建隔离的数据库会话.

我做了一个最小的,完整的例子来突出问题,请注意test_user_schema1()并且test_user_schema2()是相同的.

文件名: test_db.py

from models import User

def test_user_schema1(session):
    person_name = 'Fran Clan'
    uu = User(name=person_name)
    session.add(uu)
    session.commit()

    assert uu.id==1
    assert uu.name==person_name

def test_user_schema2(session):
    person_name = 'Stan Clan'
    uu = User(name=person_name)
    session.add(uu)
    session.commit()

    assert uu.id==1
    assert uu.name==person_name

如果db在我的测试之间真正隔离,那么两个测试都应该通过.但是,最后一次测试总是失败,因为我还没有找到使数据库会话正确回滚的方法.

conftest.py根据我在Alex Michael的博客文章中看到的内容使用以下内容,但是这个夹具代码中断了,因为它显然没有隔离夹具之间的db会话.

@pytest.yield_fixture(scope='function')
def session(app, db):
    connection = db.engine.connect()
    transaction = connection.begin()

    #options = dict(bind=connection, binds={})
    options = dict(bind=connection)
    session = db.create_scoped_session(options=options)

    yield session

    # Finalize test here
    transaction.rollback()
    connection.close()
    session.remove()

出于这个问题的目的,我构建了一个要点 …

我有以下like用于搜索博客的查询.如果我这样做,我不确定自己是否容易受到SQL注入攻击.SQLAlchemy如何处理这个问题？安全吗？

search_results = Blog.query.with_entities(Blog.blog_title).filter(Blog.blog_title.like("%"+ searchQuery['queryText'] +"%")).all()

我正在创建一个新的烧瓶应用程序使用flask-sqlalchemy和flask-restful with Python 3.4.我已经定义了我的用户模型:

from mytvpy import db
from sqlalchemy.ext.declarative import declared_attr


class BaseModel(db.Model):
    __abstract__ = True

    id = db.Column(db.Integer, primary_key=True)
    created = db.Column(db.TIMESTAMP, server_default=db.func.now())
    last_updated = db.Column(db.TIMESTAMP, server_default=db.func.now(), onupdate=db.func.current_timestamp())

    @declared_attr
    def __tablename__(cls):
        return cls.__name__


class User(BaseModel):

    username = db.Column(db.String(255), unique=True)
    password = db.Column(db.String(255))
    email = db.Column(db.String(255), unique=True)

    def __init__(self, username, password, email):
        super(User, self).__init__()
        self.username = username
        self.password = password
        self.email = email

如果我尝试在ipython中查询,它可以工作:

In [15]: from mytvpy.models.base import User

In [16]: User.query.all()
Out[16]: [<mytvpy.models.base.User at 0x7fac65b1c6a0>]

但是,如果我尝试从端点点击它:

class …

我一直努力在flask-admin中实现的一个功能是当用户编辑表单时,一旦设置了字段1就限制字段2的值.

让我用文字举一个简化的例子(实际的用例更复杂).然后我将展示实现该示例的完整要点,减去"约束"功能.

假设我们有一个数据库,可以跟踪一些软件"配方",以各种格式输出报告.在recipe我们的样本数据库的表中有两个配方:"严重的报告","ASCII艺术".

为了实现每个配方,我们从几种方法中选择一种.method我们数据库的表有两个方法:"tabulate_results","pretty_print".

每种方法都有参数.该methodarg表有两个参数名称"tabulate_results"("rows","display_total")和两个参数"pretty_print"("embellishment_character","lines_to_jump").

现在,对于每个配方("严肃报告","ASCII艺术"),我们需要提供各自方法的参数值("tabulate_results","pretty_print").

对于每个记录,该recipearg表允许我们选择一个配方(即字段1,例如"严重报告")和参数名称(即字段2).问题是显示了所有可能的参数名称,而它们需要根据字段1的值进行约束.

我们可以实现哪些过滤/约束机制,一旦我们选择"严重报告",我们知道我们将使用"tabulate_results"方法,因此只有"rows"和"display_total"参数可用？

我正在考虑一些AJAX巫术,它检查Field 1并为Field 2值设置查询,但不知道如何继续.

您可以通过使用要点来看到这一点:单击Recipe Arg选项卡.在第一行("严重报告")中,如果您尝试通过单击来编辑"Methodarg"值,则所有四个参数名称都可用,而不是仅两个.

# full gist: please run this

from flask import Flask
from flask_admin import Admin
from flask_admin.contrib import sqla
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

# Create application
app = Flask(__name__)

# Create dummy secrey key so we can use sessions
app.config['SECRET_KEY'] = '123456790'

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///a_sample_database.sqlite'
app.config['SQLALCHEMY_ECHO'] = …

我正在研究烧瓶中的小休息api.Api具有注册请求并生成单独线程以在后台运行的路由.这是代码:

def dostuff(scriptname):
    new_thread = threading.Thread(target=executescript,args=(scriptname,))
    new_thread.start()

线程启动但是当我尝试从executioncript函数插入db时它会出错.它抱怨db对象没有在应用程序中注册.

我正在动态创建我的应用程序(使用api作为Blueprint).

这是应用程序的结构

-run.py ## runner script
-config
   -development.py
   -prod.py
-app
  -__init__.py
  - auth.py
  - api_v1
     - __init__.py
     - routes.py
     - models.py

这是我的跑步者脚本run.py:

from app import create_app, db

if __name__ == '__main__':
    app = create_app(os.environ.get('FLASK_CONFIG', 'development'))
    with app.app_context():
        db.create_all()
    app.run()

以下是app/__init__.py创建应用程序的代码:

from flask import Flask, jsonify, g
from flask.ext.sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app(config_name):
    """Create an application instance."""
    app = Flask(__name__)

    # apply configuration
    cfg = os.path.join(os.getcwd(), …

我正在尝试以编程方式构建搜索查询,为此,我正在加入一个表.

class User(db.Model):
    id = db.Column(db.Integer(), primary_key=True)

class Tag(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id'))
    title = db.Column(db.String(128))
    description = db.Column(db.String(128))

这是一个人为的例子 - 我希望它有意义.

说我的搜索功能看起来像:

def search(title_arg, desc_arg):
    query = User.query
    if title_arg:
        query = query.join(Tag)
        query = query.filter(Tag.title.contains(title_arg))
    if desc_arg:
        query = query.join(Tag)
        query = query.filter(Tag.description.contains(desc_arg))

    return query

以前,我一直跟踪已经在列表中加入的表,如果表在列表中,假设它已经加入,只需添加过滤器.

如果我可以查看查询对象,看看它Tag已经加入了,那就太酷了,如果是的话就跳过它.我有一些更复杂的查询构建,真正从中受益.

如果对于我错过的搜索构建查询的策略完全不同,那也会很棒.或者,如果上面的代码没问题,如果我两次加入表格,这也是很好的信息.任何帮助都非常感激!!!

我正在使用SQLAlchemy和Pycharm,但PyCharm无法看到SQLAlchemy的方法用于自动完成功能.

码:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.sqlite3'
db = SQLAlchemy(app)


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(16), index=True, unique=True)

if __name__ == '__main__':
    db.create_all()
    app.run(debug=True)

例如,如果我想调用User类的SQLAlchemy方法,我必须手动输入完整的方法名称User.query.filter_by(username ='peter').first() 自动完成示例

如何为SQLAlchemy进行自动完成工作？

1)是的,项目被多次重新加载
2)是的,文件中的右侧翻译| 设置| 项目
3)是的,PyCharm没有处于省电模式
4)是的,我有专业版.