标签: flask-sqlalchemy

我最近开始将 SQLite 数据库移植到 PostGreSQL，以用于使用 SQLAlchemy 构建的 Flask 站点。我在 PGSQL 中有我的模式，甚至将数据插入到数据库中。但是，我无法运行常用的 INSERT 命令来向数据库添加信息。通常，我使用 SQL Alchemy 插入新记录，将 ID 列保留为 NULL，然后仅设置其他列。但是，这会导致以下错误：

sqlalchemy.exc.IntegrityError: (psycopg2.IntegrityError) null value in column "id" violates not-null constraint
DETAIL:  Failing row contains (null, 2017-07-24 20:40:37.787393+00, 2017-07-24 20:40:37.787393+00, episode_length_list = [52, 51, 49, 50, 83]

sum_length = 0

for ..., 0, f, 101, 1, 0, 0, , null).
 [SQL: 'INSERT INTO submission (date_created, date_modified, code, status, correct, assignment_id, course_id, user_id, assignment_version, version, url) VALUES (CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, %(code)s, %(status)s, %(correct)s, %(assignment_id)s, %(course_id)s, …

我正在尝试检测何时创建或删除多对多关系，但我无法找出要侦听的正确事件。

假设我有一个User模型和一个Team模型，并且我使用成员资格表来定义哪些用户属于哪些团队。这是一个简单的模式（使用 Flask-SQLAlchemy 的基本模型）：

membership_table = db.Table('membership', db.metadata,
    db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
    db.Column('team_id', db.Integer, db.ForeignKey('team.id')),
    db.PrimaryKeyConstraint('user_id', 'team_id'))

class User(db.Model):
    __tablename__ = 'user'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64))
    teams = db.relationship('Team', secondary=membership_table, backref='users')

class Team(db.Model):
    __tablename__ = 'team'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64))

我想要做的是检测某人何时加入或离开团队，并触发一个可以利用此信息执行某些操作的事件（例如在某处发送通知：“戴夫加入洋基队”；“莎拉离开大都会队”）。

我首先尝试将事件直接附加after_insert到after_delete成员资格表，但这只是失败并出现AttributeError: after_delete异常，这或多或少是我所期望的，因为它不是像其他模型一样的模型。

我尝试将set侦听器附加到User.teams：

@event.listens_for(User.teams, 'set')
def membership_updated(target, value, oldvalue, initiator):
    # compare `oldvalue` to `newvalue` to determine membership change

但是，当我在团队中添加或删除用户时，此事件根本不会触发 …

我正在尝试从两个表中查询......我的约会和客户。客户号码是我在约会中的外键，我可以从客户数据库中提取它。

现在，我只是返回客人看看它在做什么，我收到此错误：TypeError：repr返回非字符串（类型元组）

 @app.route('/calendar')
def weeklycal():
    weekyrnum=request.args.get('weekyr')
    guests = db.session.query(Appointments,Clients).filter(Appointments.clientnumber == Clients.clientnumber).filter(Appointments.weekyr == weekyrnum).all()
return 

render_template(calbyWeek.html",guests=guests)

如何查询约会和客户中的所有内容，其中 clientnumber 是要加入的列（在约会模型中定义为外键），并按周进行过滤？

我正在尝试使用可变数组来通过 SQLAlchemy 和 Postgres 存储数据，并编写了这些类：

class MutableList(Mutable, list):
    def append(self, value):
        list.append(self, value)
        self.changed()

    def pop(self, index=0):
        value = list.pop(self, index)
        self.changed()
        return value

    @classmethod
    def coerce(cls, key, value):
        if not isinstance(value, MutableList):
            if isinstance(value, list):
                return MutableList(value)
            return Mutable.coerce(key, value)
        else:
            return value


class Lead(db.Model):
    __tablename__ = 'leads'

    ...
    starred = db.Column(MutableList.as_mutable(db.ARRAY(db.Integer)))

我可以更新加星标的属性，从其中添加和弹出项目。我的问题是如何搜索数组中具有特定项目的行。这是我的查询：

leads = Lead.query.order_by(Lead.post_date.desc()).filter(Lead.starred.contains(current_user.id)).all()

但这给了我以下错误：ARRAY.contains() not implemented for the base ARRAY type; please use the dialect-specific ARRAY type

我在这里缺少什么？

基本上我想做的是，加入 2 个表“用户”和“公司”，并获取用户及其相关的公司详细信息。

这是用户模型：

class User(db.Model):
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
firstname = db.Column(db.String(10), nullable=False)
lastname = db.Column(db.String(25), nullable=False)
username = db.Column(db.String(250), nullable=False)
email = db.Column(db.String(100), nullable=False)
password = db.Column(db.String(250), nullable=False)
isPasswordReset = db.Column(db.Boolean, nullable=False)
companyId = db.Column(db.Integer, db.ForeignKey(
    'company.id'), nullable=True)

这是架构

class UserSchema(ma.Schema):
id = fields.Integer()
firstname = fields.String(required=True)
lastname = fields.String(required=True)
username = fields.String(required=True)
email = fields.String(required=True)
password = fields.String(required=True)
isPasswordReset = fields.Boolean(required=True)
companyId = fields.Integer()
company_name = fields.Nested(CompanySchema)

这是公司模型：

class Company(db.Model):
__tablename__ = 'company' …

我是新使用 sqlAlchemy 并且在创建新表时遇到问题，特别是当它有 2 个外键指向 1 个表时：

class Offers(db.Model):
    __tablename__ = 'offers'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    contact_ign = db.Column(db.String(100))
    conversion_rate = db.Column(db.Float)
    stock = db.Column(db.Integer)
    create_date = db.Column(db.DateTime(timezone=True), default=func.now())
    currency_pair = db.relationship('CurrencyPairs', backref='pair', lazy='dynamic')

class CurrencyPairs(db.Model):
    __tablename__ = 'currency_pairs'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    league = db.Column(db.String(100))
    pair_id = db.Column(db.Integer, db.ForeignKey('offers.id'))
    want = db.relationship('Currency', backref='want', lazy='dynamic')
    have = db.relationship('Currency', backref='have', lazy='dynamic')

class Currency(db.Model):
    __tablename__ = 'currency'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    poe_trade = db.Column(db.Integer, nullable=False) …

我正在尝试使用 Flask-sqlalchemy 查询动态创建 order_by 语句。

这是对我的 PurchaseOrder 方法的调用：

pos = PurchaseOrder.get_all_pos(column_order=['id', 'due_date'])

column_order 可以有一个或多个列，我们用来对查询结果进行排序。

class PurchaseOrder(PurchaseOrderDB):

    @classmethod
    def get_all_pos(cls, **kwargs):
        columns = kwargs.pop('column_order', '')
        columns_order = [getattr(cls, column) for column in columns]
        if len(columns_order) != 0:
            pos = cls.query.order_by(columns_order).all()
        else:
            pos = cls.query.all()
        return pos

columns_order 是 InstrumentedAttributes 的列表，当我将该列表传递给 order_by() 方法时，我得到了预期的以下错误：

sqlalchemy.exc.ArgumentError: SQL expression object or string expected, got object of type <class 'list'> instead

那么，当我们不知道传入多少列时，是否有其他方法可以动态创建 order_by 呢？在这种情况下，查询应如下所示：

pos = cls.query.order_by(cls.id, cls.due_date).all()

我正在编写一个烧瓶应用程序。我有两个文件。main.py 和databases.py。我想从database.py 文件创建一个数据库。main.py 应访问databases.py 文件并创建名为“Users”的数据库和表。但它显示导入错误。帮我解决这个问题

主要.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from databases import User

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

if __name__ == '__main__':
    db.create_all()
    app.run(host='0.0.0.0', port=5001)  

数据库.py

from main import db
from passlib.apps import custom_app_context as pwd_context

class User(db.Model) :
    __tablename__ = 'users'
    user_id     =   db.Column(db.Integer, primary_key = True)
    username    =   db.Column(db.String(32), index = True)
    password    =   db.Column(db.String(128))
    def hash_password(self, password) :
        self.password =pwd_context.hash(password)
    def verify_password(self, password) :
        return pwd_context.verify(password, self.password)

追溯： …

我的代码有问题，尝试使用 python 在 Flask 中设置 SQLAlchemy 数据库。代码：

谢谢您的帮助。

尝试重新安装 SQLAlchemy。

我使用的是公司笔记本电脑，应该不会有问题吧？

from flask import Flask, render_template, url_for
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

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

class Todo(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String(200), nullable=False)
    date_created = db.Column(db.DateTime, default=datetime

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

@app.route('/', methods=['POST', 'GET'])
def index():
    return render_template('index.html')

if __name__ == "__main__":
    app.run(debug=True)
(env) C:\Users\rodrigs\Documents\PythonFlaskApp>app.py
C:\Users\rodrigs\Documents\PythonFlaskApp\env\lib\site-packages\flask_sqlalchemy\__init__.py:835: FSADeprecationWarning: SQLALCHEMY_TRACK_MODIFICATIONS adds significant overhead and will be disabled by …

一个菜鸟困惑的问题：

如何以及在哪里设置 pool_size？

我发现在文档中将池大小设置为：

engine = create_engine('postgresql://me@localhost/mydb',
                       pool_size=20, max_overflow=0)

，但是有没有办法为下面的代码片段设置 pool_size ，如果是的话很高兴知道如何？

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@localhost/testdb'
db = SQLAlchemy(app)


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return '<User %r>' % self.username
    ......
    ......

错误如下：

sqlalchemy.exc.TimeoutError: QueuePool limit of size 10 overflow 10 reached, connection timed out, timeout 30