按需 SQL Alchemy ORM 表

Question

遇到一些人，希望能得到一些想法/帮助。 我有一个具有树结构的数据库，其中叶子可以作为外键参与多个父级。典型的例子是一个城市，它属于国家，属于大陆。不用说国家和大陆不应该是可重复的，因此在添加另一个城市之前，我需要在数据库中找到一个对象。如果它不存在，我必须创建它，但是如果国家不存在，那么我必须检查大陆，如果这个大陆不存在，那么我必须有它的创建过程。 到目前为止，如果我从单个文件运行它，我已经创建了一大堆项目，但是如果我将 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 中，这对我来说没有任何意义。

有什么想法吗？ 干杯

Answer 1

想在此处发布解决方案，以供其他人牢记。显然，如果模块中的类没有被使用，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 循环它们，并在必要时创建表方法。

希望对其他人有所帮助。 问候