可以在 sqlalchemy 中缓存表以在特定用例中不查询数据库吗？答案

【问题标题】：possible to cache a table in sqlalchemy to not query database in specific use case?可以在 sqlalchemy 中缓存表以在特定用例中不查询数据库吗？
【发布时间】：2022-01-09 04:05:56
【问题描述】：

我有一个用例，我正在运行一个 python 脚本，不断更新数据库中表上的特定列。脚本如下：

while True:
    events = load_events()
    for event in events:
        team_1 = event.team_1
        team_2 = event.team_2
        dd = datetime.strptime(event.gametime, "%Y-%m-%dT%H:%M:%S.000Z")
        begin = dd - timedelta(hours=4)
        end = dd + timedelta(hours=4)

        db_event = session.query(SportEvent).filter(and_(
            SportEvent.home_team.contains(event.team_1),                                      
            SportEvent.away_team.contains(event.team_2),
            SportEvent.game_time.between(begin, end))).first()
        if db_event.e_id != event.event_id:
            setattr(db_event, 'e_id', event.event_id)
    session.commit()

我的问题是每个循环花费的时间远远超过可接受的时间，因为我正在更新 100k 个事件。我想在每个循环开始时实现某种快照或缓存，因此对SportEvent 的每个查询实际上不会对数据库产生查询，而只是检查缓存。（我理解这样做的风险，并且对于我的用例来说，如何整体更新/访问该表是可以接受的）。我怎样才能实现这样的东西？

class SportEvent(Base):
    __tablename__ = 'sport_events'

    id = Column(String, primary_key=True)
    sport_id = Column(String)
    league_id = Column(String)
    away_team = Column(String)
    home_team = Column(String)
    game_time = Column(DateTime)
    dk_id = Column(String)
    fd_id = Column(String)
    cs_id = Column(String)
    bm_id = Column(String)

    def __init__(self, home_team=None,
                 away_team=None,
                 game_time=None, sport_id=None, league_id=None):
        self.id = uuid.uuid4()
        self.home_team = home_team
        self.away_team = away_team
        self.game_time = game_time
        self.sport_id = sport_id
        self.league_id = league_id
        self.dk_id = ""
        self.fd_id = ""
        self.cs_id = ""
        self.bm_id = ""

【问题讨论】：

一共有多少个团队？
@flakes 团队总数
你也可以发布SportEvent的Table类吗？需要查看列上的类型
更新事件 ID 时的标称情况是什么？通常在 100k 中更新的 event_id 的百分比是多少？
@flakes 已更新以包括 SportEvent 我还包括了一个额外的日期时间过滤器（最初是为了简化我的问题，但由于这会影响查询性能，所以我将其添加回来）

标签： python sqlalchemy

【解决方案1】：

因此，您在这里遇到的是从服务到数据库的每次往返成本。根据需要返回的记录数，有一些选项。

请注意，如果不深入了解所提供的数据，这些查询可能不是最佳的。

最明显的一种是批量查询。假设更新时间都在同一时间。我们可以这样做：

begin = None
end = None
for event in events:
    gametime = datetime.strptime(event.gametime, "%Y-%m-%dT%H:%M:%S.000Z")
    if not begin or gametime < begin:
        begin = gametime
    if not end or gametime > end:
        end = gametime 
begin = begin - timedelta(hours=4)
end = end + timedelta(hours=4)

records = session.query(SportEvent).filter(
    SportEvent.game_time.between(begin, end)
).all()

现在循环时你可以查阅你已经查询过的数据。

for event in events:
    team_1 = event.team_1
    team_2 = event.team_2
    dd = datetime.strptime(event.gametime, "%Y-%m-%dT%H:%M:%S.000Z")
    begin = dd - timedelta(hours=4)
    end = dd + timedelta(hours=4)

    db_event = next(
        (
            record
            for record in records
            if (
                team_1 in record.home_team
                and team_2 in record.away_team
                and begin < record.game_time < end
            )
        ),
        None
    )
    if db_event and db_event.e_id != event.event_id:
        db_event.e_id = event.event_id

session.commit()

同样，更好的答案需要更多地了解事件数据以及如何对其进行优化以限制返回的结果。假设只有特定的球队参加比赛 - 你可以为每个球队做一个批量查询和一个循环。

基本上，您需要查看数据中的共性并尝试尽可能少地调用数据库。

另请注意，大多数数据库对查询中的参数数量都有最大限制，对于 PostgresQL，它是 64,000，低于您的总查询要求。

【讨论】：