#!/usr/bin/env python
# coding: utf-8
import os
import sqlite3
import time
from collections import OrderedDict
from gzip import GzipFile
from hashlib import sha1
from io import BytesIO
from mimetypes import guess_type
from os import stat
from weakref import proxy
from zlib import compress
from watchdog.events import FileSystemEventHandler
from ..utils import gmt_date, log_info
from .base import CacheBackendBase
suffixes = (".py", ".pyc", ".pyo", ".so")
[文档]
class FileEventHandler(FileSystemEventHandler):
def __init__(self, app):
FileSystemEventHandler.__init__(self)
self._app = proxy(app)
self._monitored_files = set() # 记录已监控的文件
self._last_reload_time = 0 # 上次重载时间
self._reload_cooldown = 1.0 # 重载冷却时间（秒）
def _should_reload(self, path):
"""
判断文件变化是否需要重新加载应用
Args:
path: 文件路径
Returns:
True 表示需要重新加载，False 表示不需要
"""
# 排除不需要触发重载的文件
# 日志文件
if path.endswith('.log'):
return False
# Python 编译文件
if path.endswith(('.pyc', '.pyo', '.so')):
return False
# 临时文件和备份文件（包括 ~, .bak, .tmp, .swp 等）
if path.endswith(('~', '.bak', '.tmp', '.swp', '.orig')):
return False
# 检查文件名是否包含波浪号（某些编辑器创建的备份文件）
if '~' in os.path.basename(path):
return False
# __pycache__ 目录
if '/__pycache__/' in path or path.startswith('__pycache__/'):
return False
# .git 目录
if '/.git/' in path or path.startswith('.git/'):
return False
# .svn 目录
if '/.svn/' in path or path.startswith('.svn/'):
return False
# .hg 目录
if '/.hg/' in path or path.startswith('.hg/'):
return False
# node_modules 目录
if '/node_modules/' in path:
return False
# Python 源文件变化需要重新加载
if path.endswith('.py'):
return True
# 配置文件变化需要重新加载
if path.endswith(('.yaml', '.yml', '.json', '.ini', '.conf')):
return True
return False
def _can_reload_now(self):
"""
检查现在是否可以触发重载（防止频繁重载）
Returns:
True 表示可以触发重载，False 表示需要等待
"""
import time
current_time = time.time()
if current_time - self._last_reload_time >= self._reload_cooldown:
self._last_reload_time = current_time
return True
return False
[文档]
def add_monitored_file(self, path):
"""
添加需要监控的文件
Args:
path: 文件路径
"""
self._monitored_files.add(path)
def _reload_app(self, event_type, path):
"""
重新加载应用
Args:
event_type: 事件类型
path: 文件路径
"""
# 检查是否可以触发重载（防止频繁重载）
if not self._can_reload_now():
return
log_info(self._app.logger, "File %s: %s, reloading application..." % (event_type, path))
# 标记需要重新加载
self._app._need_reload = True
# 清除所有缓存
self._app.caches.data.clear()
# 关闭服务器，触发重新加载
if hasattr(self._app, 'server') and self._app.server:
try:
self._app.server.server_close()
except Exception:
pass
[文档]
def on_moved(self, event):
src_path = event.src_path
dest_path = event.dest_path
# 对于移动事件，需要特殊处理
# 如果是 .py 文件移动到备份文件（如 quickstart.py -> quickstart.py~），不触发重载
# 这是编辑器保存文件时的常见行为
src_is_backup = src_path.endswith('~') or '~' in os.path.basename(src_path)
dest_is_backup = dest_path.endswith('~') or '~' in os.path.basename(dest_path)
# 如果源文件或目标文件是备份文件，不触发重载
if src_is_backup or dest_is_backup:
return
# 对于不需要触发重载的文件，不记录日志，避免循环触发
if not self._should_reload(src_path) and not self._should_reload(dest_path):
return
log_info(self._app.logger, "%s has been moved to %s" % (src_path, dest_path))
# 判断是否需要重新加载应用
if self._should_reload(src_path) or self._should_reload(dest_path):
self._reload_app('moved', dest_path)
else:
# 只清空缓存
self._app.caches.data.clear()
[文档]
def on_created(self, event):
src_path = event.src_path
# 对于不需要触发重载的文件，不记录日志，避免循环触发
if not self._should_reload(src_path):
return
log_info(self._app.logger, "%s has been created" % src_path)
# 判断是否需要重新加载应用
if self._should_reload(src_path):
self._reload_app('created', src_path)
else:
# 只清空缓存
self._app.caches.data.clear()
[文档]
def on_modified(self, event):
src_path = event.src_path
# 对于不需要触发重载的文件，不记录日志，避免循环触发
if not self._should_reload(src_path):
return
log_info(self._app.logger, "%s has been modified" % src_path)
# 判断是否需要重新加载应用
if self._should_reload(src_path):
self._reload_app('modified', src_path)
else:
# 只清空缓存
self._app.caches.data.clear()
[文档]
def on_deleted(self, event):
src_path = event.src_path
# 对于不需要触发重载的文件，不记录日志，避免循环触发
if not self._should_reload(src_path):
return
log_info(self._app.logger, "%s has been deleted" % src_path)
# 判断是否需要重新加载应用
if self._should_reload(src_path):
self._reload_app('deleted', src_path)
else:
# 只清空缓存
self._app.caches.data.clear()
[文档]
class LiteFile(object):
def __init__(self, path, base, name, text, status_code=200):
self.status_code = int(status_code)
self.path = path
self.text = text
self.etag = sha1(text).hexdigest()
self.zlib_text = zlib_text = compress(text, 9)[2:-4]
self.zlib_etag = sha1(zlib_text).hexdigest()
stream = BytesIO()
with GzipFile(fileobj=stream, mode="wb") as f:
f.write(text)
self.gzip_text = gzip_text = stream.getvalue()
self.gzip_etag = sha1(gzip_text).hexdigest()
self.last_modified = gmt_date(stat(path).st_mtime)
mimetype, coding = guess_type(name)
headers = [("Content-Type", "text/html;charset=utf-8")]
if mimetype is not None:
headers = [("Content-Type", "%s;charset=utf-8" % mimetype)]
headers.append(("Last-Modified", self.last_modified))
self.headers = headers
[文档]
def handler(self, request):
environ = request.environ
if_modified_since = environ.get("HTTP_IF_MODIFIED_SINCE")
if if_modified_since == self.last_modified:
return request._response(304)
if_none_match = environ.get("HTTP_IF_NONE_MATCH")
accept_encodings = environ.get("HTTP_ACCEPT_ENCODING", "").split(",")
accept_encodings = [s.strip().lower() for s in accept_encodings]
headers = list(self.headers)
if "gzip" in accept_encodings:
if if_none_match == self.gzip_etag:
return request._response(304)
headers.append(("Etag", self.gzip_etag))
headers.append(("Content-Encoding", "gzip"))
text = self.gzip_text
elif "deflate" in accept_encodings:
if if_none_match == self.zlib_etag:
return request._response(304)
headers.append(("Etag", self.zlib_etag))
headers.append(("Content-Encoding", "deflate"))
text = self.zlib_text
else:
if if_none_match == self.etag:
return request._response(304)
headers.append(("Etag", self.etag))
text = self.text
headers.append(("Content-Length", "%d" % len(text)))
return request._response(self.status_code, headers=headers, content=text)
[文档]
class TreeCache(CacheBackendBase):
def __init__(self, clean_period=60, expiration_time=3600):
self.data = {}
self.clean_time = time.time()
self.clean_period = clean_period
self.expiration_time = expiration_time
self.conn = sqlite3.connect(":memory:", check_same_thread=False)
self.conn.text_factory = str
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS cache (
key VARCHAR PRIMARY KEY,
timestamp INTEGER
);
CREATE INDEX idx_cache ON cache (key);
""")
def __len__(self):
return len(self.data)
[文档]
def put(self, key, val, expiration=None):
current_time = time.time()
# 检查是否需要清理
if current_time - self.clean_time >= self.clean_period:
self.auto_clean()
# 使用传入的 expiration 或默认的 expiration_time
effective_expiration = expiration if expiration is not None else self.expiration_time
timestamp = int(current_time)
if key not in self.data:
self.conn.execute(
"""
INSERT INTO cache (key, timestamp) VALUES (?, ?);
""",
(key, timestamp),
)
else:
self.conn.execute(
"""
UPDATE cache SET timestamp=? WHERE key=?;
""",
(timestamp, key),
)
# 存储 [值, 时间戳, 过期时间] 三元组
self.data[key] = [val, timestamp, effective_expiration]
[文档]
def get(self, key):
current_time = time.time()
# 检查是否需要清理
if current_time - self.clean_time >= self.clean_period:
self.auto_clean()
ret = self.data.get(key)
if ret is None:
self.stats.record_miss()
return None
val, timestamp, expiration = (
ret[0], ret[1], ret[2] if len(ret) > 2 else self.expiration_time
)
if int(current_time - timestamp) > expiration:
del self.data[key]
self.stats.record_miss()
return None
self.stats.record_hit()
return val
[文档]
def delete(self, key):
current_time = time.time()
# 检查是否需要清理
if current_time - self.clean_time >= self.clean_period:
self.auto_clean()
curr = self.conn.execute(
"""\
SELECT key FROM cache WHERE key=? OR key LIKE ?;
""",
(key, key + "/%"),
)
keys = curr.fetchall()
self.conn.executemany(
"""\
DELETE FROM cache WHERE key=?;
""",
keys,
)
for item in keys:
item_key = item[0]
if item_key in self.data:
del self.data[item_key]
[文档]
def auto_clean(self):
"""优化的清理机制
避免每次清理都查询数据库，只在清理周期到达时执行清理
"""
current_time = time.time()
# 检查是否达到清理周期，避免频繁清理
if current_time - self.clean_time < self.clean_period:
return
# 使用批量删除，避免先查询再删除（默认过期时间）
self.conn.execute(
"DELETE FROM cache WHERE timestamp < ?;",
(int(current_time - self.expiration_time),)
)
# 清理内存缓存，考虑每条记录的独立过期时间
expired_keys = [
k for k, v in self.data.items()
if current_time - v[1] > (v[2] if len(v) > 2 else self.expiration_time)
]
for key in expired_keys:
del self.data[key]
# 更新清理时间
self.clean_time = current_time
[文档]
def exists(self, key):
"""检查键是否存在且未过期"""
return self.get(key) is not None
[文档]
def clear(self):
"""清空所有缓存"""
self.data.clear()
self.conn.execute("DELETE FROM cache")
self.clean_time = time.time()
[文档]
def close(self):
"""关闭数据库连接"""
try:
if self.conn:
self.conn.close()
except Exception:
pass
[文档]
class MemoryCache(CacheBackendBase):
def __init__(self, max_size=10000):
self._max_size = int(max_size)
self._cache = OrderedDict()
self._expiry = {} # key -> expires_at (time.time() + expiration)
def __str__(self):
return str(self._cache)
def __len__(self):
return len(self._cache)
def _is_expired(self, key):
"""Check if a key has expired."""
expires_at = self._expiry.get(key)
if expires_at is not None and time.time() > expires_at:
self.delete(key)
return True
return False
def _cleanup_expired(self):
"""Remove all expired entries (lazy cleanup)."""
now = time.time()
expired_keys = [k for k, expires_at in self._expiry.items() if now > expires_at]
for key in expired_keys:
self._cache.pop(key, None)
del self._expiry[key]
[文档]
def put(self, key, val, expiration=None):
if key in self._cache:
del self._cache[key]
elif len(self._cache) >= self._max_size:
# Try cleanup before evicting LRU
self._cleanup_expired()
if len(self._cache) >= self._max_size:
evicted_key, _ = self._cache.popitem(last=False)
self._expiry.pop(evicted_key, None)
self._cache[key] = val
if expiration is not None:
self._expiry[key] = time.time() + expiration
else:
self._expiry.pop(key, None)
[文档]
def get(self, key):
if self._is_expired(key):
self.stats.record_miss()
return None
val = self._cache.get(key)
if val is None:
self.stats.record_miss()
return None
self._cache.move_to_end(key)
self.stats.record_hit()
return val
[文档]
def delete(self, key):
self._cache.pop(key, None)
self._expiry.pop(key, None)
[文档]
def exists(self, key):
"""检查键是否存在且未过期"""
if self._is_expired(key):
return False
return key in self._cache
[文档]
def clear(self):
"""清空所有缓存"""
self._cache.clear()
self._expiry.clear()
[文档]
def expire(self, key, expiration):
"""设置键的过期时间
Args:
key: 缓存键
expiration: 过期时间（秒）
Returns:
是否设置成功
"""
if key not in self._cache or self._is_expired(key):
return False
self._expiry[key] = time.time() + expiration
return True
[文档]
def ttl(self, key):
"""获取键的剩余过期时间
Returns:
剩余过期时间（秒），键不存在返回 -2，无过期时间返回 -1
"""
if key not in self._cache or self._is_expired(key):
return -2
expires_at = self._expiry.get(key)
if expires_at is None:
return -1
remaining = expires_at - time.time()
return max(0, int(remaining))
[文档]
def close(self):
"""关闭（内存缓存无需操作）"""
pass