标签: flask-sqlalchemy

我一直试图在过去的7个小时内为我的Flask应用程序配置Alembic迁移系统但没有成功.关于我做错了什么的任何想法？

我已经尝试了我在这里找到的所有解决方案,但它们似乎对我不起作用.

I get the following error after running the commands:
$ heroku run alembic revision -m "please work" --autogenerate
Running `alembic revision -m please work --autogenerate` attached to terminal... up, run.6050
Traceback (most recent call last):
  File "/app/.heroku/python/bin/alembic", line 9, in <module>
    load_entry_point('alembic==0.5.0', 'console_scripts', 'alembic')()
  File "/app/.heroku/python/lib/python2.7/site-packages/pkg_resources.py", line 337, in load_entry_point
    return get_distribution(dist).load_entry_point(group, name)
  File "/app/.heroku/python/lib/python2.7/site-packages/pkg_resources.py", line 2281, in load_entry_point
    return ep.load()
  File "/app/.heroku/python/lib/python2.7/site-packages/pkg_resources.py", line 1991, in load
    entry = __import__(self.module_name, globals(),globals(), ['__name__'])

ImportError:没有名为config的模块

另外,请注意,我正试图让它在Heroku上工作

我的env.py文件的一部分:

from alembic import …

我有一个像这样的sqlite表.其中有3列.我想用SQLAlchemy计算user_id = 1的行数吗？

class UserImage(db.Model):
        id = db.Column(db.Integer, primary_key=True)
        user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
        photo = db.Column(db.String(250))

这将返回所有行.如何修改它以获得我的预期结果.rows = db.session.query(func.count(UserImage.user_id)).标量()

非常感谢你提前.

我有一个观点:

@app.route('/', methods=['GET', 'POST'])
@app.route('/index', methods=['GET', 'POST'])
@app.route('/index/<int:page>', methods=['GET', 'POST'])
def index(page=1):
    posts = Post.query.paginate(page, 3, False).items
    return render_template('index.html', posts=posts)

代码模板:

    {% if posts %}
    <ol>
    {% for post in posts %}
        <li>{{ post.title }}</li>  | {{ post.text }} | {{ post.time }}</li>
    {% endfor %}
    </ol>
    {% else %}
    <h2>There is no posts</h2>
    {% endif %}

    {% if posts.has_prev %}<a href="{{ url_for('index', page=posts.prev_num) }}"><< Newer posts</a>{% else %}<< Newer posts{% endif %} |
    {% if posts.has_next %}<a href="{{ url_for('index', …

我正在使用flask 及其flask-restful 扩展来制作一个简单的restful-api。我无法使用用户 ID（参考）将数据填充到问题表中。

model.py 文件如下：

class User(UserMixin, SurrogatePK, Model):

    __tablename__ = 'users'
    username = Column(db.String(80), unique=True, nullable=False)
    email = Column(db.String(80), unique=True, nullable=False)
    #: The hashed password
    password = Column(db.String(128), nullable=True)
    created_at = Column(db.DateTime, nullable=False, default=dt.datetime.utcnow)

    def __init__(self, username, email, password=None, **kwargs):
        db.Model.__init__(self, username=username, email=email, **kwargs)
        if password:
            self.set_password(password)
        else:
            self.password = None

    def set_password(self, password):
        self.password = bcrypt.generate_password_hash(password)

    def check_password(self, value):
        return bcrypt.check_password_hash(self.password, value)

    def generate_auth_token(self, expiration = 600):
        s = Serializer('secret_key', expires_in = expiration)
        return s.dumps({ 'id': self.id …

当我将 SQLAlchemy 与 Oracle 结合使用时，我还必须为主键添加序列，但迁移不会自行创建序列。如何获得要创建的序列？

我曾多次尝试调整代码以使 SQLAlchemy 为主键创建 oracle 序列，但到目前为止，我还无法获得由 SQLAlchemy 创建的 Oracle 序列。到目前为止，我有一个非常简单的用户/角色设置，并且存在表，但不存在序列。它运行时没有显示任何错误。

Model 类如下所示：

class Role(SurrogatePK, Model):
    """A role for a user."""

    __tablename__ = 'roles'
    id = db.Column(db.Integer, db.Sequence(__tablename__ + '_id_seq'), primary_key=True)
    name = Column(db.String(80), unique=True, nullable=False)
    user_id = reference_col('users', nullable=True)
    user = relationship('User', backref='roles')

    def __init__(self, name, **kwargs):
        """Create instance."""
        db.Model.__init__(self, name=name, **kwargs)

我正在使用 Flask 和 SQLAlchemy，并且在我运行之后；

$ python manage.py db init
Creating directory <lots removed here>...done

$ $ python manage.py db migrate
INFO  [alembic.runtime.migration] Context impl …

使用属性和混合变压器对遍历以下树有一点帮助；

class Category(db.Model):
    __tablename__ = 'category'
    id = db.Column(db.Integer, primary_key=True)
    parent = db.Column(db.Integer, db.ForeignKey('category.id'), nullable=True)
    name = db.Column(db.String(400), index=True, unique=True)

    children = db.relationship("Category", cascade='all, delete-orphan', backref=db.backref("child", remote_side=[id]))
    parents = db.relationship("Category", cascade='all', backref=db.backref("back", remote_side=[id]))
    entries = db.relationship("Feeds", backref='entry', lazy='dynamic')


class Feeds(db.Model):
    __tablename__ = 'feeds'
    id = db.Column(db.Integer, primary_key=True)
    category_id = db.Column(db.Integer, db.ForeignKey('category.id'))
    name = (db.String(400), index=True, unique=True)

    @property
     def parents(self):
         allparents=[]
         p = self.children
         while p:
             allparents.append(p)
             p = p.children
         return allparents

我有一个简单的对象

catlist = db.session.query(Category).filter_by(id=1).all()

1. 如何遍历所有树以获得具有可变树深度的子级，即祖先 -> 父级 -> 子级 …

我正在尝试对相关对象进行以下过滤的不同项目的SQL Alchemy查询，其等效于以下查询：

SELECT distinct items.item_id, items.item_name
FROM items
INNER JOIN categories as cat on items.category_id = cat.category_id
INNER JOIN stores on cat.store_id = stores.store_id
WHERE store.store_id = 123    

我已经创建了以下包含外键的模型，但是当我在下面运行查询时，它不能正确过滤。

items_query = (db.session.query(Store, Item)
               .filter(Store.store_id == 123)
               ).all()



#SQL Alchemy Models 
class Store(db.Model):
    __tablename__ = 'stores'
    store_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    store_name = db.Column(db.String(100), unique=True, nullable=False)

    def __repr__(self):
        return '<Store>'+str(self.store_name)

class Category(db.Model):
    __tablename__ = 'categories'
    category_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    category_name = db.Column(db.String(100), unique=True, nullable=False)
    store_id = db.Column(db.Integer, db.ForeignKey('stores.store_id'))
    store = …

我有一个使用Flask-SQLAlchemy连接到数据库的Flask应用程序。

我需要检查的行是否name='reza'存在...

我想我可以使用Any()，Exists()。

我正在尝试在heroku上部署Flask应用程序，该应用程序在Celery中使用后台任务。我已经实现了应用程序工厂模式，以便将celery进程不绑定到flask应用程序的任何一个实例。

这在本地有效，但我还没有看到错误。但是，当部署到heroku时，总是会出现相同的结果：celery任务（我只使用一个）在第一次运行时成功，但是对该任务的所有后续celery调用均失败sqlalchemy.exc.DatabaseError: (psycopg2.DatabaseError) SSL error: decryption failed or bad record mac。如果我重新启动芹菜工人，循环将继续。

有多个 问题表明此错误，但没有一个问题指定正确的解决方案。最初，我认为实现应用程序工厂模式将阻止此错误的出现，但还不完全如此。

在app/__init__.py我创建celery和db对象中：

celery = Celery(__name__, broker=Config.CELERY_BROKER_URL)
db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    db.init_app(app)
    return app

我的flask_celery.py文件创建了实际的Flask应用程序对象：

import os
from app import celery, create_app

app = create_app(os.getenv('FLASK_CONFIG', 'default'))
app.app_context().push()

我用以下命令开始芹菜： celery worker -A app.flask_celery.celery --loglevel=info

实际的芹菜任务如下所示：

@celery.task()
def task_process_stuff(stuff_id):
    stuff = Stuff.query.get(stuff_id)
    stuff.processed = True
    db.session.add(stuff)
    db.session.commit()
    return stuff

通过以下方式调用：

task_process_stuff.apply_async(args=[stuff.id], countdown=10)

图书馆版本

• 烧瓶0.12.2 …

我有两个表(location和country)试图查询; 在我的烧瓶应用中由下面的模型表示

from sqlalchemy import Column, DateTime, ForeignKey, Integer, \
                       Numeric, SmallInteger, String, Table
from sqlalchemy.orm import relationship
from sqlalchemy.schema import FetchedValue
from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()


class Country(db.Model):
    __tablename__ = 'country'

    country_id = db.Column(db.Integer, primary_key=True)
    country_name = db.Column(db.String(30), nullable=False)
    full_country_name = db.Column(db.String(90), nullable=False)
    country_code = db.Column(db.String(4), nullable=False)

    def __str__(self):
        return '%s' % self.country_name

    def __repr__(self):
        return '<Country %r>' % self.country_name


class Location(db.Model):
    __tablename__ = 'location'

    location_id = db.Column(db.Integer, primary_key=True)
    location_name = db.Column(db.String(75), nullable=False) …