SQLAlchemy 自动映射：性能最佳实践

问题描述 投票：0回答：5

我正在围绕现有（mysql）数据库构建一个Python应用程序，并使用自动映射来推断表和关系：

    base = automap_base()

    self.engine = create_engine(
        'mysql://%s:%s@%s/%s?charset=utf8mb4' % (
            config.DB_USER, config.DB_PASSWD, config.DB_HOST, config.DB_NAME
        ), echo=False
    )

    # reflect the tables
    base.prepare(self.engine, reflect=True)

    self.TableName = base.classes.table_name

使用这个我可以做诸如

session.query(TableName)
之类的事情...... 但是，我担心性能，因为每次应用程序运行时，它都会再次进行整个推理。

  • 这是合理的担忧吗？
  • 如果是这样，是否有可能“缓存”Automap 的输出？
sqlalchemy
5个回答
7
投票

性能可能是一个合理的担忧。如果数据库架构未更改，则每次运行脚本时反映数据库可能会非常耗时。这更多是开发过程中的问题，而不是启动长时间运行的应用程序的问题。如果您的数据库位于远程服务器上（同样，特别是在开发期间），这也可以节省大量时间。

我使用与答案类似的代码here（如@ACV所述）。一般计划是第一次执行反射，然后pickle元数据对象。下次运行脚本时，它将查找 pickle 文件并使用它。该文件可以在任何地方，但我将我的放在 

~/.sqlalchemy_cache
中。这是基于您的代码的示例。

import os
from sqlalchemy.ext.declarative import declarative_base

self.engine = create_engine(
    'mysql://%s:%s@%s/%s?charset=utf8mb4' % (
        config.DB_USER, config.DB_PASSWD, config.DB_HOST, config.DB_NAME
    ), echo=False
)

metadata_pickle_filename = "mydb_metadata"
cache_path = os.path.join(os.path.expanduser("~"), ".sqlalchemy_cache")
cached_metadata = None
if os.path.exists(cache_path):
try:
    with open(os.path.join(cache_path, metadata_pickle_filename), 'rb') as cache_file:
        cached_metadata = pickle.load(file=cache_file)
except IOError:
    # cache file not found - no problem, reflect as usual
    pass

if cached_metadata:
    base = declarative_base(bind=self.engine, metadata=cached_metadata)
else:
    base = automap_base()
    base.prepare(self.engine, reflect=True) # reflect the tables

    # save the metadata for future runs
    try:
        if not os.path.exists(cache_path):
            os.makedirs(cache_path)
        # make sure to open in binary mode - we're writing bytes, not str
        with open(os.path.join(cache_path, metadata_pickle_filename), 'wb') as cache_file:
            pickle.dump(Base.metadata, cache_file)
    except:
        # couldn't write the file for some reason
        pass

self.TableName = base.classes.table_name

对于使用声明性表类定义的任何人，假设 

Base
对象定义为例如

Base = declarative_base(bind=engine)

metadata_pickle_filename = "ModelClasses_trilliandb_trillian.pickle"

# ------------------------------------------
# Load the cached metadata if it's available
# ------------------------------------------
# NOTE: delete the cached file if the database schema changes!!
cache_path = os.path.join(os.path.expanduser("~"), ".sqlalchemy_cache")
cached_metadata = None
if os.path.exists(cache_path):
    try:
        with open(os.path.join(cache_path, metadata_pickle_filename), 'rb') as cache_file:
            cached_metadata = pickle.load(file=cache_file)
    except IOError:
        # cache file not found - no problem
        pass
# ------------------------------------------

# define all tables
#
class MyTable(Base):
    if cached_metadata:
        __table__ = cached_metadata.tables['my_schema.my_table']
    else:
        __tablename__ = 'my_table'
        __table_args__ = {'autoload':True, 'schema':'my_schema'}

...
# ----------------------------------------
# If no cached metadata was found, save it
# ----------------------------------------
if cached_metadata is None:
    # cache the metadata for future loading
    # - MUST DELETE IF THE DATABASE SCHEMA HAS CHANGED
    try:
        if not os.path.exists(cache_path):
            os.makedirs(cache_path)
        # make sure to open in binary mode - we're writing bytes, not str
        with open(os.path.join(cache_path, metadata_pickle_filename), 'wb') as cache_file:
            pickle.dump(Base.metadata, cache_file)
    except:
        # couldn't write the file for some reason
        pass

重要提示!!如果数据库架构发生更改，您必须删除缓存文件以强制代码自动加载并创建新的缓存。如果不这样做，更改将反映在代码中。这是一件很容易忘记的事情。

6
投票

我认为“反映”数据库的结构并不是正确的方法。除非您的应用程序尝试从结构中“推断”事物，就像对源文件进行静态代码分析一样，否则这是不必要的。在运行时反映它的另一个原因是减少使用 SQLAlchemy 开始“使用”数据库的时间。然而：

另一种选择是使用 SQLACodegen (https://pypi.python.org/pypi/sqlacodegen):

它将“反映”您的数据库一次，并创建一组准确度为 99.5% 的声明性 SQLAlchemy 模型供您使用。但是，这确实要求您随后使模型与数据库结构保持同步。我认为这不是一个大问题，因为您已经使用的表足够稳定，因此它们结构的运行时反映不会对您的程序产生太大影响。

生成声明性模型本质上是反射的“缓存”。只是 SQLACodegen 将其保存到一组非常可读的类 + 字段中，而不是内存中的数据。即使结构发生变化，并且我自己对生成的声明性模型进行了“更改”，以后每当我进行数据库更改时，我仍然会在项目中使用 SQLACodegen。这意味着我的模型彼此之间通常是一致的，并且我没有由于复制粘贴而出现打字错误和数据不匹配等问题。

4
投票

你的第一个问题的答案很大程度上是主观的。您正在添加数据库查询以获取应用程序加载时间的反射元数据。该开销是否很大取决于您的项目要求。

作为参考，我在工作中有一个使用反射模式的内部工具，因为加载时间对于我们的团队来说是可以接受的。如果它是面向外部的产品，情况可能并非如此。我的预感是，对于大多数应用程序来说，反射开销不会主导应用程序的总加载时间。

如果您认为这对您的目的很重要，那么这个问题有一个有趣的答案，用户会腌制数据库元数据以便在本地缓存它。

1
投票

除此之外，@Demitri 的回答接近正确，但是（至少在

sqlalchemy 1.4.29
中），当从缓存文件生成时，该示例将在最后一行
self.TableName = base.classes.table_name
失败。在这种情况下 
declarative_base()
没有属性 
classes
。

修复就像更改一样简单：

if cached_metadata:
    base = declarative_base(bind=self.engine, metadata=cached_metadata)
else:
    base = automap_base()
    base.prepare(self.engine, reflect=True) # reflect the tables

到

if cached_metadata:
    base = automap_base(declarative_base(bind=self.engine, metadata=cached_metadata))
    base.prepare()
else:
    base = automap_base()
    base.prepare(self.engine, reflect=True) # reflect the tables

这将创建具有适当属性的 

automap
对象。

0
投票

对于希望在 SQLAlchemy 2.0+ 中执行此操作的人，我能够使其与下面的代码一起使用，以实现与 @dimitri

的其他答案中建议的类似功能 
import os
import pickle
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import create_engine

def load_metadata(cache_path, filename):
    """Load metadata from a pickle file."""
    try:
        with open(os.path.join(cache_path, filename), 'rb') as cache_file:
            return pickle.load(file=cache_file)
    except (IOError, EOFError):
        return None

def save_metadata(metadata, cache_path, filename):
    """Save metadata to a pickle file."""
    try:
        if not os.path.exists(cache_path):
            os.makedirs(cache_path)
        with open(os.path.join(cache_path, filename), 'wb') as cache_file:
            pickle.dump(metadata, cache_file)
    except IOError as e:
        print(f"Error saving metadata: {e}")

def reflect_tables(engine, cached_metadata=None):
    """Reflect the tables and return the base class."""
    if cached_metadata:
        base = automap_base(metadata=cached_metadata)
        base.prepare()
    else:
        base = automap_base()
        base.prepare(autoload_with=engine)
    return base

# Main execution
DATABASE_URL = os.getenv("DATABASE_URL")
engine = create_engine(DATABASE_URL)

metadata_pickle_filename = "mydb_metadata.pkl"
cache_path = "sqlalchemy_cache"
cached_metadata = load_metadata(cache_path, metadata_pickle_filename)

base = reflect_tables(engine, cached_metadata)

if not cached_metadata:
    # Save metadata if it was not loaded from cache
    save_metadata(base.metadata, cache_path, metadata_pickle_filename)
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.