forked from username/flaskpaste
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
"""API blueprint registration."""
|
|
|
|
import threading
|
|
import time
|
|
|
|
from flask import Blueprint, current_app
|
|
|
|
bp = Blueprint("api", __name__)
|
|
|
|
# Thread-safe cleanup scheduling
|
|
_cleanup_lock = threading.Lock()
|
|
_cleanup_times: dict[str, float] = {
|
|
"pastes": 0.0,
|
|
"hashes": 0.0,
|
|
"rate_limits": 0.0,
|
|
"audit": 0.0,
|
|
"short_urls": 0.0,
|
|
}
|
|
_CLEANUP_INTERVALS = {
|
|
"pastes": 3600, # 1 hour
|
|
"hashes": 900, # 15 minutes
|
|
"rate_limits": 300, # 5 minutes
|
|
"audit": 86400, # 24 hours
|
|
"short_urls": 3600, # 1 hour
|
|
}
|
|
|
|
|
|
def reset_cleanup_times() -> None:
|
|
"""Reset cleanup timestamps. For testing only."""
|
|
with _cleanup_lock:
|
|
for key in _cleanup_times:
|
|
_cleanup_times[key] = 0.0
|
|
|
|
|
|
@bp.before_request
|
|
def run_scheduled_cleanup():
|
|
"""Periodically run cleanup tasks on schedule."""
|
|
now = time.time()
|
|
|
|
with _cleanup_lock:
|
|
# Cleanup expired pastes
|
|
if now - _cleanup_times["pastes"] >= _CLEANUP_INTERVALS["pastes"]:
|
|
_cleanup_times["pastes"] = now
|
|
from app.database import cleanup_expired_pastes
|
|
|
|
count = cleanup_expired_pastes()
|
|
if count > 0:
|
|
current_app.logger.info(f"Cleaned up {count} expired paste(s)")
|
|
|
|
# Cleanup expired content hashes
|
|
if now - _cleanup_times["hashes"] >= _CLEANUP_INTERVALS["hashes"]:
|
|
_cleanup_times["hashes"] = now
|
|
from app.database import cleanup_expired_hashes
|
|
|
|
count = cleanup_expired_hashes()
|
|
if count > 0:
|
|
current_app.logger.info(f"Cleaned up {count} expired hash(es)")
|
|
|
|
# Cleanup rate limit entries
|
|
if now - _cleanup_times["rate_limits"] >= _CLEANUP_INTERVALS["rate_limits"]:
|
|
_cleanup_times["rate_limits"] = now
|
|
from app.api.routes import cleanup_rate_limits
|
|
|
|
count = cleanup_rate_limits()
|
|
if count > 0:
|
|
current_app.logger.info(f"Cleaned up {count} rate limit entries")
|
|
|
|
# Cleanup old audit logs
|
|
if now - _cleanup_times["audit"] >= _CLEANUP_INTERVALS["audit"]:
|
|
_cleanup_times["audit"] = now
|
|
if current_app.config.get("AUDIT_ENABLED", True):
|
|
from app.audit import cleanup_old_audit_logs
|
|
|
|
count = cleanup_old_audit_logs()
|
|
if count > 0:
|
|
current_app.logger.info(f"Cleaned up {count} old audit log entries")
|
|
|
|
# Cleanup expired short URLs
|
|
if now - _cleanup_times["short_urls"] >= _CLEANUP_INTERVALS["short_urls"]:
|
|
_cleanup_times["short_urls"] = now
|
|
from app.database import cleanup_expired_short_urls
|
|
|
|
count = cleanup_expired_short_urls()
|
|
if count > 0:
|
|
current_app.logger.info(f"Cleaned up {count} expired short URL(s)")
|
|
|
|
|
|
from app.api import routes # noqa: E402, F401
|