标签: flask-sqlalchemy

我试图按照这个例子在一个使用Python Enum类型的表中有一个枚举列.我定义了枚举,然后将其传递给列,如示例所示,但我明白了ValueError: <enum 'FruitType'> is not a valid Enum.如何使用Python枚举正确定义SQLAlchemy枚举列？

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import enum

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

class FruitType(enum.Enum):
    APPLE = "Crunchy apple"
    BANANA = "Sweet banana"

class MyTable(db.Model):
    id = db.Column(db.Integer, primary_key = True)
    fruit_type = db.Column(enum.Enum(FruitType))

File "why.py", line 32, in <module>
    class MyTable(db.Model):
  File "why.py", line 34, in MyTable
    fruit_type = db.Column(enum.Enum(FruitType))
  File "/usr/lib/python2.7/dist-packages/enum/__init__.py", line 330, in __call__
    return cls.__new__(cls, value)
  File "/usr/lib/python2.7/dist-packages/enum/__init__.py", line …

我在我的网络项目中使用SQlAlchemy.我应该使用scoped_session(session_maker())或plain session_maker()以及为什么？或者我应该使用其他东西？

## model.py
from sqlalchemy import *
from sqlalchemy.orm import *

engine = create_engine('mysql://dbUser:dbPassword@dbServer:dbPort/dbName',
pool_recycle=3600, echo=False)
metadata = MetaData(engine)
Session = scoped_session(sessionmaker())
Session.configure(bind=engine)
user = Table('user', metadata, autoload=True)

class User(object):
pass

usermapper = mapper(User, user)

## some other python file called abc.py
from models import *

def getalluser():
   session = Session()  
   session.query(User).all()
   session.flush()
   session.close()

## onemore file defg.py
from models import *

def updateuser():
   session = Session()  
   session.query(User).filter(User.user_id == '4').update({User.user_lname: 'villkoo'})
   session.commit()
   session.flush()
   session.close()

我为每个请求创建一个session = Session()对象,然后关闭它.我做对了还是有更好的方法吗？

我有一个简单的用户模型,定义如下:

# models.py
from datetime import datetime
from myapp import db

class User(db.Model):
  id = db.Column(db.Integer(), primary_key=True)
  email = db.Column(db.String(100), unique=True)
  password = db.Column(db.String(100))
  date_updated = db.Column(db.DateTime())

  def __init__(self, email, password, date_updated=None):
    self.email = email
    self.password = password
    self.date_updated = datetime.utcnow()

当我创建一个新的User对象时,我的date_updated字段被设置为当前时间.我想做的是这样做,每当我保存对User对象的更改时,我的date_updated字段将自动设置为当前时间.

我已经搜索了文档,但对于我的生活,我似乎无法找到任何参考.我对SQLAlchemy很新,所以我真的没有任何经验可以借鉴.

会喜欢一些反馈,谢谢.

我正在尝试使用SQLAlchemy模块在python中编写批量upsert(而不是在SQL中!).

我在SQLAlchemy添加上收到以下错误:

sqlalchemy.exc.IntegrityError: (IntegrityError) duplicate key value violates unique constraint "posts_pkey"
DETAIL:  Key (id)=(TEST1234) already exists.

我有一个posts在id列上使用主键调用的表.

在这个例子中,我已经在db中有一行id=TEST1234.当我尝试db.session.add()使用id设置为的新帖子对象时TEST1234,我得到上面的错误.我的印象是,如果主键已经存在,记录将会更新.

如何仅使用基于主键的Flask-SQLAlchemy进行升级？有简单的解决方案吗？

如果没有,我总是可以检查并删除任何匹配id的记录,然后插入新记录,但这对我的情况来说似乎很昂贵,我不希望有很多更新.

我最近更新了 Flask-SQLAlchemy，现在db.create_all正在提高RuntimeError: working outside of application context. 我怎么打电话create_all？

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

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

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

db.create_all()

这会引发以下错误：

Traceback (most recent call last):
  File "/home/david/Projects/flask-sqlalchemy/example.py", line 11, in <module>
    db.create_all()
  File "/home/david/Projects/flask-sqlalchemy/src/flask_sqlalchemy/extension.py", line 751, in create_all
    self._call_for_binds(bind_key, "create_all")
  File "/home/david/Projects/flask-sqlalchemy/src/flask_sqlalchemy/extension.py", line 722, in _call_for_binds
    engine = self.engines[key]
  File "/home/david/Projects/flask-sqlalchemy/src/flask_sqlalchemy/extension.py", line 583, in engines
    app = current_app._get_current_object()  # type: ignore[attr-defined] …

我有两个Flask-SQLAlchemy模型,具有简单的一对多关系,如下面的最小示例:

class School(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30))
    address = db.Column(db.String(30))

class Teacher(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30))
    id_school = db.Column(db.Integer, db.ForeignKey(School.id))

    school = relationship('School', backref='teachers')

然后我向使用该关系的教师添加混合属性,如:

@hybrid_property
def school_name(self):
    return self.school.name

当我使用它时,该属性工作得很好teacher_instance.school_name.但是,我也想提出类似的查询Teacher.query.filter(Teacher.school_name == 'x'),但这给了我一个错误:

`AttributeError: Neither 'InstrumentedAttribute' object nor 
'Comparator' object has an attribute 'school_name'`. 

在SQLAlchemy文档之后,我添加了一个简单的混合表达式,如下所示:

@school_name.expression
def school_name(cls):
    return School.name

但是,当我再次尝试相同的查询时,它会生成一个没有join子句的SQL查询,因此我获得了School中所有可用的行,而不仅仅是那些与Teacher中的外键匹配的行.

从SQLAlchemy文档中我意识到表达式需要一个已经存在连接的上下文,所以我再次尝试查询:

Teacher.query.join(School).filter(Teacher.school_name == 'x')

这实际上是有效的,但是如果我需要学校模型的知识来实现​​这一目的,那就首先试图在那里试图获得语法糖.我希望有一种方法可以在表达式中加入,但我无法在任何地方找到它.该文档有一个示例,表达式返回一个直接用它构建的子查询select(),但即使这对我没用.

有任何想法吗？

UPDATE

在下面的Eevee回答之后,我使用了建议的关联代理并且它有效,但我也对它应该与select()子查询一起工作的评论感到好奇,并试图弄清楚我做错了什么.我最初的尝试是:

@school_name.expression
def school_name(cls):
    return select(School.name).where(cls.id_school == School.id).as_scalar() …

这是我的Flask-SQLAlchemy声明代码:

from sqlalchemy.ext.associationproxy import association_proxy
from my_flask_project import db


tagging = db.Table('tagging',
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id', ondelete='cascade'), primary_key=True),
    db.Column('role_id', db.Integer, db.ForeignKey('role.id', ondelete='cascade'), primary_key=True)
)


class Tag(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)

    def __init__(self, name=None):
        self.name = name

    @classmethod
    def delete_orphans(cls):
        for tag in Tag.query.outerjoin(tagging).filter(tagging.c.role_id == None):
            db.session.delete(tag)


class Role(db.Model):

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id', ondelete='cascade'))
    user = db.relationship('User', backref=db.backref('roles', cascade='all', lazy='dynamic'))
    ...
    tags = db.relationship('Tag', secondary=tagging, cascade='all', backref=db.backref('roles', cascade='all'))
    tag_names = association_proxy('tags', 'name')

    __table_args__ …

SQLAlchemy one和first方法之间有什么区别

我正在尝试将Flask-SQLAlchemy模型分成单独的文件.当我试着奔跑时,db.create_all()我得到了No application found. Either work inside a view function or push an application context.

shared/db.py:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

app.py:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from shared.db import db

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'My connection string'
db.init_app(app)

user.py:

from shared.db import db

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String(300), unique=True, nullable=False)
    password = db.Column(db.Text, nullable=False)

我一直在学习 Flask Web 应用程序框架并且对它感到很舒服。我以前构建了一个简单易用的应用程序，它运行良好。我在同一个项目上工作，但尝试使用 TDD 来实现它。我遇到了以前从未见过的数据库错误，不知道如何修复。

当我检查我的代码时，我看不出任何问题。它看起来也与工作项目的代码相同，所以我真的不知道我做错了什么。

这是错误：

(env) PS C:\coding-projects\task-master-tdd> flask shell
Python 3.8.5 (tags/v3.8.5:580fbb0, Jul 20 2020, 15:43:08) [MSC v.1926 32 bit (Intel)] on win32
App: project [development]
Instance: C:\coding-projects\task-master-tdd\instance
>>> from project import db
>>> db
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "c:\coding-projects\task-master-tdd\env\lib\site-packages\flask_sqlalchemy\__init__.py", line 1060, in __repr__
    self.engine.url if self.app or current_app else None
  File "c:\coding-projects\task-master-tdd\env\lib\site-packages\flask_sqlalchemy\__init__.py", line 943, in engine
    return self.get_engine()
  File "c:\coding-projects\task-master-tdd\env\lib\site-packages\flask_sqlalchemy\__init__.py", line 962, in get_engine
    return connector.get_engine()
  File …