标签: flask-sqlalchemy

我确实有 2 个这样的课程 =>

class User(db.Model):
    __tablename__ = "user"
    user_id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), unique=True, nullable=False)
    password = db.Column(db.String(77), unique=False, nullable=False)
    server_limit = db.Column(db.Integer, unique=False, nullable=False, server_default="4")
    servers = db.relationship('Server', backref='owner', lazy='dynamic')

class Server(db.Model):
    __tablename__ = "server"
    server_id = db.Column(db.Integer, primary_key=True)
    server_admin = db.Column(db.Integer, db.ForeignKey("user.user_id"))
    server_port = db.Column(db.Integer, unique=False, nullable=False)
    server_slot = db.Column(db.Integer, unique=False, nullable=False, server_default="32")

现在我试图获取 server_slot 列的总和，例如 user_id 为 1。

我知道对此有已接受的答案，但不同之处在于我尝试使用服务器来做到这一点（db.relationship我在用户类中分配的服务器（）来完成此操作。

我用为 User 类创建的替代方法来做到这一点=>

def used(self):
    return db.session.execute("SELECT SUM(server.server_slot) FROM server WHERE server_admin={}".format(self.user_id)).scalar()

我怎样才能做到这一点 …

我正在使用 Flask、Flask-Socketio 和 Flask-SQLAlchemy 创建实时报告应用程序。我当前的设计在连接上创建一个后台线程，用于查询 API 并插入到应用程序数据中。但是，当运行这个时，我收到错误

RuntimeError: No application found. Either work inside a view function or push an application context.

Flask_react_app.py：

from threading import Lock
from flask import Blueprint, render_template
from .model import Stock
from . import socketio

main = Blueprint('main', __name__)
thread_lock = Lock()
thread = None


@main.route('/')
def index():
    """Serve client-side application."""
    return render_template('index.html', async_mode=socketio.async_mode)


def generate_data():
    """
    returns the Stock object query set
    """
    return [i.serialize for i in Stock.query.all()]


def background_thread():
    """Example of how …

我对数据库设计和网页设计还很陌生，所以请耐心等待。

我有一个 Flask 应用程序，我需要使用 Flask-sqlalchemy 根据 url 中传递的参数在数据库中进行搜索，例如：

.../index?part_number=123*321$description1=&description2=TestText

星号将被解释为一个或多个字符，** 将被解释为单个字符。所以我将用 % 和 _ 替换它们，并终止任何特殊字符。

我的问题是：在数据库中，有些字段将为空字符串，有些字段将为 Null。但我需要将它们都解释为空字符串，以便当 arg 为空或 * 时返回它们。

我知道我可以做一些类似的事情.like(part_number == None | part_number == '%')。但我只想在搜索字符串为 * 的情况下执行此操作，但是当我有 10 个不同的参数，全部与一起组合在一起时，如何以巧妙的方式执行此操作？

这是相关代码的片段和精简版本。我不太确定如何制作一个可以自行运行以进行测试的小程序。

filter_args = ['part_number', 'description1', 'description2']  # actual code has ~10 args
filters = dict()
for arg in filter_args:
  filter_str = request.args.get(arg, type=str, default='*')  # get filter arg from url

  filter_str = '*' if filter_str == '' else filter_str  # if the filter string is empty, …

我有一个 Flask 应用程序，使用 Flask-SQLAlchemy 和 MySQL 数据库，其中数据库定义如下：

db.py:

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

main.py:

from db import db
app = Flask(__name__)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+pymysql://" + \
        DB_USERNAME + ":" + DB_PASSWORD + "@" + DB_HOST + "/" + DB_DATABASE
db.init_app(app)

@app.teardown_appcontext
def teardown_db(error):
    db.session.close()
    db.engine.dispose()

user.py:

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)

我使用模型查询数据库，db.engine.execute()在需要时使用编写原始 SQL 查询或使用集成的 Flask-SQLAlchemy API 来读取数据，例如User.query.filter_by().all(). …

我正在移植我使用 mysql 库来查询 mysql 的旧代码。我试图使用flask_SQLALchemy 来做同样的事情，但无法找到查询“或”的替代选项。

下面是更新后的代码片段，介绍了它的工作原理。

from flask_sqlalchemy import SQLAlchemy                                                                                                                                                        
from myapp import APP                                                                                                                                                                          
from myapp.models.nodes import Host                                                                                                                                                            


DB = SQLAlchemy(APP)  
# config is taken care here                                                                                                                                                                         

APP.app_context().push()                                                                                                                                                                       
with APP.app_context():                                                                                                                                                                        
    DB.init_app(APP)     

Hostname_var = "some hostname"
session_query = Host.query.filter_by(hostname=hostname_var).first()

同样，我想查询下面给出的语句。我可以使用 sqlalchemy 但使用不同的库听起来并不理想。

"Select count(*) as c from table_name where ver='some version' and arch='some arch' and pool=1 and (state='Ready' or state='Provisioning');" table_name 的类称为 Host（如示例中所使用的）。

如有任何帮助，我们将不胜感激，谢谢。

我想使用烧瓶棉花糖中的架构为从模型检索的数据应用自定义日期时间格式。

目前使用的架构如下：

class ScheduleListSchema(ma.Schema):
    class Meta:
        fields = ('id', 'start_time', 'end_time')

在此start_time格式中2018-12-05T03:00:00+00:00

所以我想为架构中的值创建自定义格式start_time。

我正在尝试使用 Flask-SQLAlchemy 查询数据库中的用户个人资料页面

到目前为止我还没有解决这个问题的方法，只能使用以下方法查询所有用户数据users.query.all()

每个用户都有自己的role_id、department_id、researchfield_id。

如何通过ID查询出所有与User有关系的Role、Department、ResearchField数据？

class User(UserMixin, db.Model):

    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    password_hash = db.Column(db.String(128))
    is_admin = db.Column(db.Boolean, default=False)
    department_id = db.Column(db.Integer, db.ForeignKey('departments.id'))
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    research_id = db.Column(db.Integer, db.ForeignKey('researchfields.id'))

class Department(db.Model):
    __tablename__ = "departments"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(sqlalchemy.types.NVARCHAR(100), unique=True)
    user = db.relationship('User', backref='department',
                                lazy='dynamic')

class Role(db.Model):

    __tablename__ = 'roles'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(sqlalchemy.types.NVARCHAR(100), unique=True)
    users = db.relationship('User', backref='role',
                                lazy='dynamic')


class ResearchField(db.Model):

    __tablename__ = "researchfields"

    id = db.Column(db.Integer, …

class Contacts(db.Model):
    """
    sno, name, email, phone_num, msg, date
    """
    sno = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(20), nullable=False)
    phone_num = db.Column(db.String(12), nullable=False)
    msg = db.Column(db.String(120), nullable=False)
    date = db.Column(db.String(12), nullable=True)

回溯（最近一次调用最后）：文件“C：/Users/rithi/Flask_Blog/main.py”，第11行，在类Contacts（db.Model）中：文件“C：/Users/rithi/Flask_Blog/main.py” "，第 15 行，在联系人 sno = db.Column(db.Integer, Primary_key=True) TypeError: Column() 获得意外的关键字参数 'primary_key'

我正在使用 FlaskSqlachemy 和 Postgresql 数据库。

我有一个名为 Order 的表和一个名为 Attempt（尝试交付订单）的表，它们处于一对多关系。尝试表有一个 order.id 的 FK。

我想在同一会话中将记录插入到 Order 和 Attempt 表中。问题是我在插入之前不知道 order.id 。这是因为order.id是具有自增功能的Order表的PK。

我可以先插入订单，然后检查 order.id，最后插入尝试。我的问题是：如果第二次插入失败，我最终会得到一个没有与之关联的尝试的订单。

这种情况下的最佳做法是什么？

下面是我的数据库模型的简化版本：

class Order(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    attempts = db.relationship('Attempt', backref='order', lazy='select',
                               order_by="Attempt.date")

class Attempt(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    order_id = db.Column(None, db.ForeignKey('order.id'))

现在我的解决方案如下：

    # Create an order object
    new_order = Order()

    # Try to add it to db
    try:
        db.session.add(new_order)
        db.session.commit()
    except Exception:
        ....
        return redirect(url_for('scheduleorder'))
    else:
        order = Order.query.filter_by(token=token).first()

        # Create an attempt object
        new_attempt = …

我正在尝试在烧瓶中使用数据库

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

在其中我创建了一个基本的登录页面和一个注册页面，当我尝试使用 html 代码时：

<form action="/login" method="post">
        <input autocomplete="off" autofocus name="username" placeholder="Username" type="text">
        <input class="form-control" name="password" placeholder="Password" type="password">
    <button class="btn btn-primary" type="submit">Log In</button>
            <li style="list-style-type:none;"><a href="/register">Register</a></li>
</form>

在这个Python Flask代码上：

# This is the log in page
@app.route("/login", methods=["GET", "POST"])
def login():
    """Log user in"""

    # Forget any user_id
    session.clear()

    # User reached route via POST (as by submitting a form via POST)
    if request.method == "POST":

        # Ensure username was submitted …