使用 Python CGI 限制对外部数据库的调用

Question

我有一个 Python CGI 脚本，可以从 GPS 服务中提取数据；我希望网页上的此信息大约每 10 秒更新一次（GPS 服务的 TOS 允许的最大值）。但可能有 100 个用户同时查看该网页，所有用户都调用该脚本。

我认为用户的脚本需要从缓冲区页面获取数据，该页面本身每十秒只更新一次。如果没有人直接查看内容（并且不访问 CGI），如何使该缓冲区页面自动更新？有更好的方法来实现这一目标吗？

Answer 1

将 GPS 数据查询的结果与日期时间一起缓存在文件或数据库 (sqlite) 中。

然后，您可以根据上次缓存的日期时间进行日期时间检查，以启动另一个 GPS 数据查询。

不过，您可能会遇到 cgi 和日期时间检查的并发问题...

为了解决并发问题，您可以使用 sqlite，并将写入放在 try/ except 中。
这是使用 sqlite 的示例缓存实现。

import datetime
import sqlite3 

class GpsCache(object):
    db_path = 'gps_cache.db'
    def __init__(self):
        self.con = sqlite3.connect(self.db_path)
        self.cur = self.con.cursor()

    def _get_period(self, dt=None):
        '''normalize time to 15 minute periods'''
        if dt.minute < 15:
           minute_period = 0
        elif 15 <= dt.minute < 30:
           minute_period = 15
        elif 30 <= dt_minute < 45: 
           minute_period = 30
        elif 45 <= dt_minute:
           minute_period = 25
        period_dt = datetime.datetime(year=dt.year, month=dt.month, day=dt.day, hour=dt.hour, minute=minute_period)
        return period_dt

    def get_cache(dt=None):
        period_dt = self._get_period(dt)
        select_sql = 'SELECT * FROM GPS_CACHE WHERE date_time = "%s";' % period_dt.strftime('%Y-%m-%d %H:%M')
        self.cur.execut(select_sql)
        result = self.cur.fetchone()[0]
        return result


    def put_cache(dt=None, data=None):
        period_dt = self._get_period(dt)
        insert_sql = 'INSERT ....'  # edit to your table structure
        try:
            self.cur.execute(insert_sql)
            self.con.commit()
        except sqlite3.OperationalError:
            # assume db is being updated by another process with the current resutls and ignore
            pass

所以我们现在有了缓存工具的实现端。

您需要首先检查缓存，然后如果它不是“新鲜”（不返回任何内容），请使用当前方法获取数据。然后缓存你​​抓取的数据。
您可能应该更好地组织它，但您应该在这里了解总体思路。

使用此示例，您只需将当前对“remote_get_gps_data”的调用替换为“get_gps_data”即可。

from gps_cacher import GpsCache

def remote_get_gps_data():
    # your function here
    return data

def get_gps_data():
    data = None
    gps_cache = GpsCache()
    current_dt = datetime.datetime.now()
    cached_data = gps_cache.get_cache(current_dt)    
    if cached_data:
        data = cached_data
    else:
        data = remote_get_gps_data()
        gps_cache.put_cache(current_dt, data)
    return data