遇到一些家伙，希望得到一些想法/帮助。 我有一个具有树结构的数据库，其中 leaf 可以作为外键参与多个父项。典型的例子是一个城市，它属于国家，属于大陆。不用说，国家和大陆不应该是可重复的，因此在添加另一个城市之前，我需要在数据库中找到一个对象。如果它不存在，我必须创建它，但如果例如国家还不存在，那么我必须检查大陆，如果这个不存在，那么我必须有它的创建过程。 到目前为止，如果我从单个文件运行它，我会创建一大堆项目，但是如果我将 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)第二个问题是关于Python中OOP的声明性基础(declarative_base())的继承。我想删除一些代码重复，并将它们拉到一个类中，其他类将从中派生。例如，上面的代码可以取出到单独的函数中，并具有如下所示的内容：
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)
这里的问题是元数据试图将不相关的类链接在一起，并且需要父类中存在__tablename__和一些索引列OnDemandTables这对我来说没有任何意义。
有什么想法吗？ 干杯
想在这里发布解决方案，供团伙的其他成员牢记。显然，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 遍历它们，并在必要时创建表方法。
希望它对其他人有所帮助。 问候