标签: sqlalchemy

是否可以以类似的方式使用正则表达式session.query(MyObject).filter_by(REGEX)？

如果没有,我如何使用sqlAlchemy检索具有以某个值开头的varchar PK的记录(例如,所有那些城市字段以"SA"开头的记录)？谢谢.

初学者问题 - sqlite和sqlalchemy有什么区别？

使用SQLAlchemy 0.7.1和M​​ySQL 5.1数据库,我建立了多对多关系,如下所示:

user_groups = Table('user_groups', Base.metadata,
    Column('user_id', String(128), ForeignKey('users.username')),
    Column('group_id', Integer, ForeignKey('groups.id'))
)

class ZKUser(Base, ZKTableAudit):
    __tablename__ = 'users'

    username   = Column(String(128), primary_key=True)
    first_name = Column(String(512))
    last_name  = Column(String(512))

    groups = relationship(ZKGroup, secondary=user_groups, backref='users')

class ZKGroup(Base, ZKTableAudit):
    __tablename__ = 'groups'

    id          = Column(Integer, primary_key=True)
    name        = Column(String(512))

用户可以属于多个组,而组可以包含多个用户.

我正在尝试做的是构建一个SQLAlchemy查询,该查询仅返回属于组列表中至少一个组的用户.

我玩这个in_函数,但这似乎只适用于测试列表中成员资格的标量值.我不是一个SQL编写者,所以我甚至不知道SELECT这需要什么样的声明.

在表的SQLAlchemy类中是否有一种方法可以为该表定义/创建触发器和索引？

例如,如果我有一个基本的表...

class Customer(DeclarativeBase):
    __tablename__ = 'customers'
    customer_id = Column(Integer, primary_key=True,autoincrement=True)
    customer_code = Column(Unicode(15),unique=True)
    customer_name = Column(Unicode(100))
    search_vector = Column(tsvector) ## *Not sure how do this yet either in sqlalchemy*.

我现在想要创建一个触发器来更新"search_vector"

CREATE TRIGGER customers_search_vector_update BEFORE INSERT OR UPDATE
ON customers
FOR EACH ROW EXECUTE PROCEDURE
tsvector_update_trigger(search_vector,'pg_catalog.english',customer_code,customer_name);

然后我想将该字段添加为索引...

create index customers_search_vector_indx ON customers USING gin(search_vector);

现在我从我的应用程序执行任何类型的数据库重新生成后,我必须为tsvector列添加列,触发器定义,然后是psql的索引语句.不是世界末日,而是容易忘记一步.我全都是关于自动化的,所以如果我能在应用程序设置期间完成这一切,那么奖金!

我正在寻找创建新SQLAlchemy对象的最快方法,只有在数据库中不存在它时.

我现在这样做的方法是首先获取查询的计数以查看它是否存在,如果不存在 - 然后我创建它.例如:

if not User.query.filter(email=user.email).count():
    db.session.add(user)
    db.session.commit()

这是最好的方法吗？会喜欢一些反馈.谢谢!

我在Flask上有一个工作的Web应用程序与SqlAlchemy用于审核新闻,它有一些api方法来处理审核请求,例如批准,拒绝当前选择的新闻,列出它们等等.我想为这些方法编写单元测试,以及我让它们工作,但我不明白,如何在一个数据库会话中实现从测试用例执行的所有请求,以便我可以删除对数据库的所有更改.或者是否有其他更清洁或正确的方法来做到这一点？我发现可能我需要的只是SqlAlchemy中的"scoped_session",但实现它的所有工作都失败了.如果这是正确的方法,请告诉我在哪里使用这行代码(在设置中,或在测试用例set_up方法中).

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
session_factory = sessionmaker()
Session = scoped_session(session_factory) 

我正在使用SQLAlchemy connection.execute(sql)将选择结果转换为映射数组.有以下代码

def __sql_to_data(sql):
    result = []
    connection = engine.connect()
    try:
        rows = connection.execute(sql)
        for row in rows:
            result_row = {}
            for col in row.keys():
                result_row[str(col)] = str(row[col])
            result.append(result_row)
    finally:
        connection.close()
    return result

例如

__sql_to_data(sql_get_scan_candidate)

return __sql_to_data(sql_get_profile.format(user_id))

问题 如何修改程序以使类似的东西成为可能

return __sql_to_data(sql_get_profile,user_id)

在我安装Mariadb 10之后,Mysql工作台和JPDB客户端都连接并正常工作,所以下一步是用Python编程(使用SQLAlchemy),这似乎需要MySQL-python所以我去更新并得到:"mysql_config not found"I看着"通常的地方",没看到档案......

所以我按照之前关于SO的问题的一些想法 并尝试安装:apt-get install libmysqlclient-dev

让我去了:以下包有未满足的依赖:libmysqlclient-dev:取决于:libmysqlclient18(= 5.5.35-0ubuntu0.13.10.2)但是要安装10.0.10 + maria-1~saucy

哪种打砖墙适合我

我一直在使用sqlalchemy.orm中的@validates装饰器来验证字段,只要所有字段彼此独立,所有字段都很顺利,例如:

@validates('field_one')
def validates_field_one(self, key, value):
   #field one validation

@validates('field_two')
def validates_field_two(self, key, value):
   #field two validation

但是,现在我需要进行一些验证,这需要同时访问field_one和field_two.看起来验证接受验证装饰器的多个参数,但是,它只会为每个参数运行一次验证函数,如下所示:

@validates('field_one', 'field_two')
def validates_fields(self, keys, values):
   #field validation

结果验证field_one的工作流程,然后验证field_two.但是,我想同时验证两者(一个简单的例子是断言field_one的值不是field_two的值,其中一个示例是禁止在图中的自循环,其中field_one和field_two引用节点,它正在边缘上执行验证).如何才能成为最佳方式？

我有一个正在运行的Flask应用程序,它是根据我们在网上和Miguel Grinberg的"Flask Web Development"一书中找到的最佳实践组合而建立的.

我们现在需要第二个Python应用程序,它不是Web应用程序,需要访问与Flask应用程序相同的模型.我们想重复使用相同的模型,因此两个应用程序都可以从共享代码中受益.

我们已经删除了对flask-sqlalchemy扩展(我们之前使用过Flask应用程序时)的依赖关系.并将其替换为此处描述的SQLalchemy Declarative扩展,这有点简单(Flask-SQLalchemy为标准SQLAlchemy添加了一些特定的东西)

根据示例,我们在根目录中创建了一个database.py文件.在我们的例子中,有两个与Declarative扩展示例不同的东西:我将引擎和会话放在一个类中,因为我们所有的模型都使用db.session而不是db_session,并且我将带有配置值的字典传递给init( ),以便我可以使用不同的配置从Flask和其他应用程序重新使用此database.py.它看起来像这样:

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base


class Database(object):

    def __init__(self, cfg):
        self.engine = create_engine(cfg['SQLALCHEMY_DATABASE_URI'], convert_unicode=True)
        self.session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=self.engine))

    class Model(object):
        pass

Base = declarative_base()

所以现在我们来解决实际问题.Flask创建一个包含配置选项的类字典对象,并将它们作为属性添加到app-instance.它从实例文件夹,站点根目录中的config.py和环境变量加载它们.我需要从Flask传递配置字典,所以我需要Flask来FIRST加载和组装配置,然后初始化数据库,并在app文件的根目录中有一个(配置的)db对象.但是,我们遵循应用工厂模式,因此我们可以针对不同情况(测试,生产,开发)使用不同的配置.

这意味着我们app/__init__.py看起来像这样(简化):

from flask import Flask
from database import Database
from flask.ext.mail import Mail
from flask_bcrypt import Bcrypt
from config import config …