我在SQLAlchemy中有两个模型，这些模型已自动加入外键，如下所示：
class Parent(Base):
__tablename__ = 'parents'
id = Column(Integer, primary_key=True)
name = Column(String(300), nullable=False)
metadata_id = Column(Integer, nullable=True, index=True)
class Child(Base):
__tablename__ = 'children'
id = Column(Integer, primary_key=True)
name = Column(String(300), nullable=False)
parent_metadata_id = \
Column(ForeignKey('parents.metadata_id'),
nullable=True, primary_key=True)
parent = relationship(u'Parent')
这个工作很好，我可以很容易地从孩子的父母那里得到帮助。现在，由于超出了此问题范围的技术原因，我不得不摆脱数据库中的外键。我试图在SQLAlchemy中解决该问题，但是没有任何替换代码（使用primaryjoins或backrefs）起作用。我在这里看到了另一个答案，只是说这是对SQLAlchemy的谎言，并告诉我我具有外键关系，但这使Alembic尝试在我自动生成的每个新修订版本上创建外键关系，这确实很烦人。什么是正确的方法？
我正在编写一个需要连接到PostgreSQL数据库和MySQL数据库的SQLAlchemy应用程序.基本上我是从现有MySQL数据库加载数据,对其进行一些转换,然后将其保存在PostgreSQL中.
我正在使用SQLAlchemy管理PostgreSQL架构declarative base.MySQL数据库已经存在,我通过SQLAlchemy的反射访问模式.两者都有非常不同的模式.
我知道我需要专门engines为每个数据库,但我不清楚我是否需要以下任何一个的专用对象:
Base - 我认为这对应于数据库架构.由于两个数据库都有非常不同的模式,因此我需要专用Base于每个模式.
Metadata - 这是否是一个包含所有引擎的所有模式的单个全局元数据对象？
Sessions - 我不确定,但我认为每个数据库需要单独的会话？或者可以单个session共享多个engine/ Base组合？我正在使用scoped_sessions.
我困惑的一部分来自不理解之间的差异Base和Metadata.在SQLAlchemy的文档说:
MetaData是一个容器对象,它将所描述的数据库(或多个数据库)的许多不同功能保持在一起.
这似乎意味着单个metadata可以容纳多个Base,但我仍然有点模糊它的工作原理.例如,我希望能够metadata.create_all()在PostgreSQL中调用和创建表,而不是MySQL.
我正在尝试将以下内容翻译成sqlalchemy:
SELECT COUNT(TableA.id) AS 'number_of_staff'
FROM TableA
WHERE datediff(month, TableA.Date, getDate()) >= 0
AND lookup_TableB = '4'
AND (TableA.End_Date IS NULL OR TableA.End_Date = '')
到目前为止,我有:
get_tablea_number = self.db_session.query(func.count(TableA.id)).\
filter(TableA.lookup_TableB == property_id,
todays_date.month - extract('month', TableA.date) >= 0,
or_(TableA.end_date == None,
TableA.end_date == '')).scalar()
然而,它输出不同的值到上面的SQL查询,我假设是下降到todays_date.month - extract('month', TableA.date) >= 0(其中todays_date.month只是一个日期时间对象)
我在第57行看到了Flask-common的以下代码片段：
id = db.Column(UUID, default=lambda: str(uuid.uuid4()), primary_key=True)
所以我想尝试一下并在我的应用中使用它models.py（因为我更喜欢uuid为id输入类型）
from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy
from sqlalchemy.dialects.postgresql import UUID
import uuid
from app import db
class CostCenter(db.Model):
__tablename__ = "costcenter"
id = db.Column('id', UUID(as_uuid=True), default=lambda: str(uuid.uuid4()), primary_key=True)
name = db.Column('name', db.Text)
def __init__(self, name):
self.name = name
def __repr__(self):
return '<id {}>'.format(self.id)
但是当我尝试运行时python manage.py db upgrade，会导致错误：
File "build/bdist.linux-x86_64/egg/sqlalchemy/sql/compiler.py", line 190, in __init__
File "build/bdist.linux-x86_64/egg/sqlalchemy/sql/compiler.py", line 213, in process
File "build/bdist.linux-x86_64/egg/sqlalchemy/sql/visitors.py", line …我有这样的代码：
thing = thing.query.filter_by(id=thing_id).limit(1).all()[0]
all()[0]在这种limit(1)情况下感觉有点混乱和多余。有没有更简洁（和/或其他最佳方式）的方法来实现这一目标？
我正在使用sqlalchemy，也使用alembic进行迁移（和flask-sqlalchemy）。我有一个使用EARTH数据类型的postgres表。
CREATE TABLE things
(
id INTEGER PRIMARY KEY NOT NULL,
name TEXT,
earth_location EARTH
)
这是我的sqlalchemy映射：
class Thing(db.Model):
__tablename__ = 'things'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.UnicodeText, nullable=False)
earth_location = db.Column(???)
如何映射接地柱？
Alembic迁移能够应付吗？
谢谢！
我有一个Tag表可以拥有Tag同一个类的父类.
我想要一个查询来返回Tag没有任何子节点的所有实例.
这是SqlAlchemy类的代码:
class Tag(db.Model):
__tablename__ = 'tags'
id = db.Column(db.String(32), primary_key=True)
name = db.Column(db.String(45),nullable=False)
subject_id = db.Column(db.Integer, db.ForeignKey('subjects.id'), nullable=False)
parent_tag_id = db.Column(db.Integer, db.ForeignKey('tags.id'), nullable=True)
subject = db.relationship('Subject', backref=db.backref('tags', lazy='dynamic'))
parent_tag = db.relationship('Tag',
remote_side=[id],
backref=db.backref('children', lazy='dynamic'))
def __init__(self, name, subject_id, parent_tag_id=None):
self.id = uuid.uuid4().hex
self.name = name
self.subject_id = subject_id
self.parent_tag_id = parent_tag_id
这是我对查询的最佳尝试:
def get_all_subject_tags_ordered():
_child_tag = aliased(Tag)
return db.session.query(Tag)\
.join(_child_tag, Tag.children)\
.filter(func.count(Tag.children) == 0)\
.filter(Tag.subject_id.isnot(None))\
.order_by(Tag.name)\
.all()
这给了我错误:
sqlalchemy.exc.ProgrammingError:(pymysql.err.ProgrammingError)(1111,u'Invalid使用组功能'的)[SQL:u'SELECT tags.id AS tags_id,tags.name …
我有下一个查询.
item = [item.export_simple()
for item in session.query(Item)
.filter(and_(
Item.companyId == company_id,
or_(
True if search == "" else None,
or_(*[Item.name.like('%{0}%'.format(s)) for s in words]),
or_(*[Item.code.like('%{0}%'.format(s)) for s in words])
))).order_by(Item.name)]
还有这个.
if type == "code":
src = [Item.code.like('%{0}%'.format(s)) for s in words]
elif type == "name":
src = [Item.name.like('%{0}%'.format(s)) for s in words]
session.query(Item)
.filter(and_(
Item.companyId == company_id,
Item.typeItem == item_type,
or_(
True if search == "" else None,
or_(*src)
)))
当前是否可以TABLESAMPLE在SQLAlchemy中的ORM查询中的某些表上使用？
我能找到的最接近的是Query.suffix_with(*suffixes)，但是将后缀放在查询的末尾，而不是表的末尾。
如果有帮助，这是我的用例：
我的查询当前是：
query = session.query(A).options(
subqueryload(A.rel1),
subqueryload(A.rel2)
).filter(A.id >= min_id, A.id < max_id, [...])
并且我想TABLESAMPLE在桌子上，A这样我就可以A使用某些过滤器有效地仅加载的子集及其对应的关系。我正在使用PostgreSQL 9.5作为数据库。
我正在通过pymssql在后端利用的库从SQL Server数据库下载一些数据.a的结果curson.execute("""<QUERY BODY>""")是一个sqlalchemy.engine.result.ResultProxy对象.如何检查查询结果是否为空,因此没有行？
cur = ff.sql.create_engine(server=dw.address, db=dw.BI_DW,
login=":".join([os.environ["SQL_USER"],
os.environ["SQL_PASSWD"]]))
for n in range(100):
result = cur.execute("""QUERY BODY;""")
if result:
break
不幸的是,当SQL查询没有返回任何行时,它result永远不会出现None.
检查的最佳方法是什么？