标签: flask-sqlalchemy

我的User模特是

class User(UserMixin, db.Model):
    __tablename__ = 'users'
    # noinspection PyShadowingBuiltins
    uuid = Column('uuid', GUID(), default=uuid.uuid4, primary_key=True,
                  unique=True)
    email = Column('email', String, nullable=False, unique=True)
    _password = Column('password', String, nullable=False)
    created_on = Column('created_on', sa.types.DateTime(timezone=True),
                        default=datetime.utcnow())
    last_login = Column('last_login', sa.types.DateTime(timezone=True),
                        onupdate=datetime.utcnow())

其中GUID是sqlalchemy docs中描述的自定义类型(完全相同)

现在我跑的时候

alembic revision --autogenerate -m "Added initial table"

我得到了我upgrade()的

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('uuid', sa.GUID(), nullable=False),
    sa.Column('email', sa.String(), nullable=False),
    sa.Column('password', sa.String(), nullable=False),
    sa.Column('created_on', …

我正在尝试使用烧瓶形式填充sqlalchemy请求结果的选择字段.

这是代码:

def possible_book():
    return Book.query.with_entities(Book.id).all()

class AuthorForm(Form):
    familyname  = TextField('familyname', [validators.Required()])
    firstname   = TextField('firstname', [validators.Required()])
    book_list = QuerySelectField('book_list',query_factory=possible_book,get_label='title',allow_blank=True)

这是模板:

 <form action="" method="post" name="edit_author" enctype="multipart/form-data">
     {{form.hidden_tag()}}
    <p>Veuillez donner son prénom (obligatoire) :    {{form.firstname(value=author.firstname)}}</p>
    <p>Nom de famille (obligatoire) : {{form.familyname(value=author.familyname)}}</p>
    <p>Livre : {{form.book_list}}</p>

    <p><input type="submit" value="Envoyer"></p>
 </form>

查看:

@app.route('/edit_author/<number>', methods = ['GET', 'POST'])
   def edit_author(number):
   if number == 'new':
      author = []
   else:
      author = Author.query.filter_by(id = number).first()
   form = AuthorForm()
   if form.validate_on_submit():
      a = Author(firstname= form.firstname.data, familyname= form.familyname.data)
      a.books = …

我正在尝试使用Flask和SQLAlchemy创建一对多的关系.

我希望一对多关系如此:

"对于任何一部电影,都可以有多个角色"

这就是我到目前为止所拥有的,但它现在正在我的数据库中保存为一对一.(一个电影到一个字符,在DB中为多个字符保存多次)

class Movie(db.Model):
    __tablename__ = "movies"
    id = db.Column('movies_id', db.Integer, primary_key=True)
    movie_type = db.Column('movie_type', db.Text())

    def __init__(self, movie_type):
        self.movie_type = movie_type

    def __repr__(self):
        return '<Movie %r>' % self.id

class Character(db.Model):
    __tablename__ = "characters"
    id = db.Column('character_id', db.Integer, primary_key=True) 
    character_description = db.Column('character_description', db.Text())

    movie_id = db.Column(db.Integer, db.ForeignKey('movies.movie_id'))
    movie = db.relationship('Movie', backref='characters', lazy='dynamic')

    def __init__(self, character_description, movie):
        self.character_description = character_description

        self.movie = movie

    def __repr__(self):
        return '<Character %r>' % self.id

我像这样保存在DB中:

movie = models.movie(movie_type)
character = models.Character(character_description, movie)

db.session.add(movie) …

我是python的新手,我刚刚学会了如何使用flask restless和flask sql-alchemy创建一个api.但是我想用随机值为数据库播种.我该如何实现这一目标？请帮忙.这是api代码......

import flask
import flask.ext.sqlalchemy
import flask.ext.restless
import datetime

DATABASE = 'sqlite:///tmp/test.db'

#Create the Flask application and the FLask-SQLALchemy object
app = flask.Flask(__name__)
app.config ['DEBUG'] = True
app.config['SQLALCHEMY_DATABASE_URI'] =  DATABASE
db = flask.ext.sqlalchemy.SQLAlchemy(app)

#create Flask-SQLAlchemy models
class TodoItem(db.Model):
    id = db.Column(db.Integer, primary_key = True)
    todo = db.Column(db.Unicode)
    priority = db.Column(db.SmallInteger)
    due_date = db.Column(db.Date)

#Create database tables
db.create_all()

#Create Flask restless api manager
manager = flask.ext.restless.APIManager(app, flask_sqlalchemy_db = db)

#Create api end points
manager.create_api(TodoItem, methods = ['GET','POST','DELETE','PUT'], results_per_page = …

使用Flask-SQLAlchemy和Foreign Key约束的工厂男孩的正确方法是什么？

请考虑以下Flask SQLAlchemy Model设置:

# coding=utf-8
from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)

# ------------------------------
#   SQLAlchemy Table Models
# ------------------------------
class User(db.Model):
    """ A SQLAlchemy simple model class who represents a user with a ForeignKey Constraint"""
    __tablename__ = 'UserTable'

    user_pk = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.Unicode(20))
    group_fk = db.Column(db.ForeignKey("GroupTable.group_pk"), nullable=False)


class Group(db.Model):
    """ A SQLAlchemy simple model class who represents a user """
    __tablename__ = 'GroupTable'

    group_pk = …

我正在努力编写py.test fixtures来管理我的应用程序数据库,以最大限度地提高速度,支持pytest-xdist测试的并行化,并将测试彼此隔离开来.

我正在对PostgreSQL 9.4数据库使用Flask-SQLAlchemy 2.1.

以下是我要完成的大致概述:

1. $ py.test -n 3 旋转三个测试会话以运行测试.

2. 在每个会话中,py.test fixture运行一次以设置事务,创建数据库表,然后在会话结束时回滚事务.创建数据库表需要在PostgreSQL事务中发生,该事务仅对该特定测试会话可见,否则由pytest-xdist原因创建的并行化测试会话会相互冲突.

3. 为每个测试运行的第二个py.test fixture连接到现有事务,以便查看创建的表,创建嵌套保存点,运行测试,然后回滚到嵌套保存点.

4. 理想情况下,这些pytest灯具支持调用的测试db.session.rollback().在这个SQLAlchemy文档的底部有一个潜在的配方来完成这个.

5. 理想情况下,pytest fixtures应该产生db对象,而不仅仅是会话,以便人们可以编写测试而不必记住使用与db.session他们在整个应用程序中使用的标准不同的会话.

这是我到目前为止所拥有的:

import pytest

# create_app() is my Flask application factory
# db is just 'db = SQLAlchemy()' + 'db.init_app(app)' within the create_app() function
from app import create_app, db as _db 


@pytest.yield_fixture(scope='session', autouse=True)
def app():
    '''Session-wide test application'''
    a = create_app('testing')
    with a.app_context():
        yield a

@pytest.yield_fixture(scope='session')
def db_tables(app):
    '''Session-wide test database''' …

association_table = Table("association_table",
                          Base.metadata,
                          Column("show_id", Integer(), ForeignKey("show_times.id"), primary_key=True),
                          Column("theater_id", Integer(), ForeignKey("theaters.id")))

association_table2 = Table("association_table2",
                           Base.metadata,
                           Column("show_id", Integer(), ForeignKey("show_times.id"), primary_key=True),
                           Column("movie_id", Integer(), ForeignKey("movies.id")))



class Movie(Base):
    __tablename__ = "movies"
    id = Column(Integer, primary_key=True)
    title = Column(String(), unique=True)
    plot = Column(String())
    duration = Column(String())
    rating = Column(String())
    trailer = Column(String())
    imdb = Column(String())
    poster = Column(String())
    summary = Column(String())

class Theater(Base):
    __tablename__ = "theaters"
    id = Column(Integer, primary_key=True)
    zip_code = Column(String())
    city = Column(String())
    state = Column(String())
    address = Column(String())
    phone_number = Column(String()) …

我正在开发一个Flask项目,我正在使用Flask-SQLAlchemy.
我需要使用多个现有的数据库.
我创建了"app"对象和SQLAlchemy对象:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy


app = Flask(__name__)
db = SQLAlchemy(app)

在配置中,我设置了默认连接和其他绑定:

SQLALCHEMY_DATABASE_URI = 'postgresql://pg_user:pg_pwd@pg_server/pg_db'
SQLALCHEMY_BINDS = {
    'oracle_bind': 'oracle://oracle_user:oracle_pwd@oracle_server/oracle_schema',
    'mssql_bind': 'mssql+pyodbc://msssql_user:mssql_pwd@mssql_server/mssql_schema?driver=FreeTDS'
}

然后我使用声明性系统创建了表模型,并在需要时设置 __bind_key__参数以指示表所在的数据库.
例如:

class MyTable(db.Model):
    __bind_key__ = 'mssql_bind'
    __tablename__ = 'my_table'

    id = db.Column(db.Integer, nullable=False, primary_key=True)
    val = db.Column(db.String(50), nullable=False)

通过这种方式,一切正常,当我进行查询时,它是在正确的数据库上进行的.

阅读SQLAlchemy文档和Flask-SQLALchemy文档我理解这些事情(我写下来检查我理解正确):

• 您可以通过会话处理事务.
• 在SQLAlchemy中,您可以将会话与特定引擎绑定.
• Flask-SQLAlchemy在请求开始时自动创建会话(scoped_session),并在请求结束时销毁它
  

所以我可以这样做:

record = MyTable(1, 'some text')
db.session.add(record)
db.session.commit()

我无法理解当我们在Flask-SqlAlchemy中使用多个关于会话的数据库时会发生什么.

我验证了系统能够通过__bind_key__参数在正确的数据库中正确绑定表,因此,我可以通过在不同的数据库中插入数据db.session,并在提交时保存所有内容.

但是,我不能理解Flask-SQLAlchemy是否创建多个会话(每个引擎一个会话)或者以不同的方式管理事物.
在这两种情况下,如何引用特定数据库的会话/事务？
如果我使用db.session.commit()系统对所有涉及的数据库进行提交,但是如果我只想为单个数据库提交,我该怎么办？
我会做的事情如下:

db.session('mssql_bind').commit()

但我无法弄清楚如何做到这一点.

我还看到了一个Flask-SQLAlchemy实现,它可以简化这些情况的管理:

问题:https …

我正在使用flask migrateflask-sqlalchemy进行数据库创建和迁移.

一切都工作正常,直到我更改我的数据库用户密码包含'@'然后它停止工作所以,我更新了我的代码基于 写密码包含特殊字符时的连接字符串

它适用于应用但不适用于烧瓶迁移,它在迁移时显示错误

即 python manage.py db migrate

ValueError: invalid interpolation syntax in u'mysql://user:p%40ssword@localhost/testdb' at position 15

密码在这里被p@ssword转义urlquote(见上面的问题链接).

完整错误堆栈:

Traceback (most recent call last):
  File "manage.py", line 20, in <module>
    manager.run()
  File "/usr/local/lib/python2.7/dist-packages/flask_script/__init__.py", line 412, in run
    result = self.handle(sys.argv[0], sys.argv[1:])
  File "/usr/local/lib/python2.7/dist-packages/flask_script/__init__.py", line 383, in handle
    res = handle(*args, **config)
  File "/usr/local/lib/python2.7/dist-packages/flask_script/commands.py", line 216, in __call__
    return self.run(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/flask_migrate/__init__.py", line 177, in migrate
    version_path=version_path, rev_id=rev_id)
  File "/usr/local/lib/python2.7/dist-packages/alembic/command.py", line 117, …

我有一个Flask项目,通过Flask-SQLAlchemy与MySQL数据库交互.

我的问题是,如何根据值或其他值从数据库中选择一行.

我在SQL中想要的结果看起来像这样

SELECT id FROM users WHERE email=email OR name=name;

如何在Flask-SQLAlchemy中实现这一目标？