SQL Alchemy ORM表按需
遇到了一些家伙，希望能得到一些想法/帮助。我有一个树形结构的数据库，其中叶子可以作为外键参与几个父级。一个典型的例子是一个城市，它属于国家和大陆。不用说，国家和大陆不应该是可重复的，因此在添加另一个城市之前，我需要在DB中找到一个对象。如果它不存在，我必须创建它，但是如果国家还不存在，那么我必须检查大陆，如果这个大陆不存在，那么我必须为它创建过程。到目前为止，如果我从单个文件中运行，我已经创建了一大堆项目，但是如果我将SQL炼金术代码推入模块中，情况就会不同。由于某些原因，元范围变得有限，如果表还不存在，那么当我查询外键存在(从国家的城市)时，代码开始抛出ProgrammingError异常。我已经拦截了它，在我正在寻找的类(国家)的__init__类构造函数中，我正在检查表是否存在，如果不存在，就创建它。有两件事我有问题，需要一个建议: 1)表的验证效率很低-我正在使用Base.metadata.sorted_tables数组，我必须通过它来查看并确定表结构是否与我的类__tablename__匹配。例如：
for table in Base.metadata.sorted_tables:
# Find a right table in the list of tables
if table.name == self.__tablename__:
if __DEBUG__:
print 'DEBUG: Found table {} that equal to the class table {}'.format(table.name, self.__tablename__)
if not table.exists():
session.get_bind().execute(table.create())不用说，这需要时间，我正在寻找更有效的方法来做同样的事情。
2)第二个问题是声明式基础(declarative_base())相对于Python的继承。我想去掉一些重复的代码，并将它们放入一个类中，其他类将从中派生。例如，上面的代码可以放入单独的函数中，并具有以下内容：
Base = declarative_base()
class OnDemandTables(Base):
__tablename__ = 'no_table'
# id = Column(Integer, Sequence('id'), nullable=False, unique=True, primary_key=True, autoincrement=True)
def create_my_table(self, session):
if __DEBUG__:
print 'DEBUG: Creating tables for the class {}'.format(self.__class__)
print 'DEBUG: Base.metadata.sorted_tables exists returns {}'.format(Base.metadata.sorted_tables)
for table in Base.metadata.sorted_tables:
# Find a right table in the list of tables
if table.name == self.__tablename__:
if __DEBUG__:
print 'DEBUG: Found table {} that equal to the class table {}'.format(table.name, self.__tablename__)
if not table.exists():
session.get_bind().execute(table.create())
class Continent(OnDemandTables):
__tablename__ = 'continent'
id = Column(Integer, Sequence('id'), nullable=False, unique=True, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
def __init__(self, session, continent_description):
if type(continent_description) != dict:
raise AttributeError('Continent should be described by the dictionary!')
else:
self.create_my_table(session)
if 'continent' not in continent_description:
raise ReferenceError('No continent can be created without a name!. Dictionary is {}'.
format(continent_description))
else:
self.name = continent_description['continent']
print 'DEBUG: Continent name is {} '.format(self.name)这里的问题是，元数据试图将不相关的类链接在一起，并要求父OnDemandTables类中存在__tablename__和一些索引列，这对我来说没有任何意义。
有什么想法吗？干杯
回答 1
Stack Overflow用户
发布于 2018-03-14 23:51:04
我想把解决方案张贴在这里，以供其他人记住。显然，如果模块中的类没有被使用，那么SQLAlchemy看不到这些类。经过几天的尝试，我发现最简单的解决方案是以半手动的方式完成--不依赖ORM为您构建和构建数据库，而是使用类方法以一种手动的方式完成这一部分。代码是：
__DEBUG__ = True
from sqlalchemy import String, Integer, Column, ForeignKey, BigInteger, Float, Boolean, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
from sqlalchemy.exc import ProgrammingError
from sqlalchemy import create_engine, schema
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
engine = create_engine("mysql://test:test123@localhost/test", echo=True)
Session = sessionmaker(bind=engine, autoflush=False)
session = Session()
schema.MetaData.bind = engine
class TemplateBase(object):
__tablename__ = None
@classmethod
def create_table(cls, session):
if __DEBUG__:
print 'DEBUG: Creating tables for the class {}'.format(cls.__class__)
print 'DEBUG: Base.metadata.sorted_tables exists returns {}'.format(Base.metadata.sorted_tables)
for table in Base.metadata.sorted_tables:
# Find a right table in the list of tables
if table.name == cls.__tablename__:
if __DEBUG__:
print 'DEBUG: Found table {} that equal to the class table {}'.format(table.name, cls.__tablename__)
if not table.exists():
if __DEBUG__:
print 'DEBUG: Session is {}, engine is {}, table is {}'.format(session, session.get_bind(), dir(table))
table.create()
@classmethod
def is_provisioned(cls):
for table in Base.metadata.sorted_tables:
# Find a right table in the list of tables
if table.name == cls.__tablename__:
if __DEBUG__:
print 'DEBUG: Found table {} that equal to the class table {}'.format(table.name, cls.__tablename__)
return table.exists()
class Continent(Base, TemplateBase):
__tablename__ = 'continent'
id = Column(Integer, Sequence('id'), nullable=False, unique=True, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
def __init__(self, session, provision, continent_description):
if type(continent_description) != dict:
raise AttributeError('Continent should be described by the dictionary!')
else:
if 'continent' not in continent_description:
raise ReferenceError('No continent can be created without a name!. Dictionary is {}'.
format(continent_description))
else:
self.name = continent_description['continent']
if __DEBUG__:
print 'DEBUG: Continent name is {} '.format(self.name)它提供了以下内容: 1.类方法is_provisioned和create_table可以在初始代码启动期间调用，并将反映数据库状态2.类继承是从保存这些方法的第二个类完成的，它不会干扰ORM类，因此不会被链接。
由于Base.metadata.sorted_tables循环的结果只是一个类表，因此即使进一步删除该循环，代码也可以进行优化。下面的操作是组织类，检查它们的表，并可能以列表的形式创建它们，同时牢记它们的链接，然后使用is_provisioned遍历它们，如果需要，还可以创建表方法。
希望能对其他人有所帮助。问候
https://stackoverflow.com/questions/49242521
复制相似问题
腾讯云开发者
