[t-015] Create staging.py — file upload staging + 30-day cleanup sch #15

New issue

Open

opened 2026-07-05 08:31:01 +00:00 by david · 0 comments

david commented 2026-07-05 08:31:01 +00:00

Owner

Copy link

Goal

Implement the file upload staging service and the lifespan-scheduled cleanup task.

References

Reference

• DESIGN.md §4 Config (staging_dir, max_upload_size_mb, staging_cleanup_days)
• AGENT.md §10 Security Constraints (File upload safety: MIME validation, 50 MB cap)

Implementation Steps

Implementation Steps

1. Create src/services/staging.py (complete implementation)

Replace placeholder with full module:

# src/services/staging.py
import asyncio
from pathlib import Path
from loguru import logger
import uuid
from src.core.config import load_config

def _allowed_mime_types() -> set[str]:
    return {
        "application/pdf", "text/plain", "text/html", "text/css",
        "text/javascript", "image/png", "image/jpeg", "image/gif",
        "image/webp", "application/json", "application/xml",
        "video/mp4", "audio/mpeg", "application/zip",
    }

def validate_upload(file_data: bytes, mime_type: str) -> bool:
    config = load_config()
    max_bytes = config.bff.max_upload_size_mb * 1024 * 1024
    if len(file_data) > max_bytes:
        return False
    if mime_type not in _allowed_mime_types():
        return False
    return True

def stage_file(file_data: bytes, name: str, mime_type: str) -> dict:
    config = load_config()
    staging_dir = Path(config.bff.staging_dir)
    staging_dir.mkdir(parents=True, exist_ok=True)
    file_id = uuid.uuid4().hex[:12]
    dest = staging_dir / f"{file_id}/{name}"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_bytes(file_data)
    return {
        "staging_path": str(dest.relative_to(staging_dir)),
        "name": name,
        "size": len(file_data),
        "mime_type": mime_type,
    }

async def cleanup_staging():
    config = load_config()
    staging_dir = Path(config.bff.staging_dir)
    if not staging_dir.exists():
        return
    cutoff_days = config.bff.staging_cleanup_days
    import time
    now = time.time()
    cutoff = now - (cutoff_days * 86400)
    for item in staging_dir.rglob("*"):
        if item.is_file() and item.stat().st_mtime < cutoff:
            item.unlink()
            logger.info(f"Cleaned up stale staging file: {item}")

async def schedule_cleanup():
    """Run cleanup every 24 hours."""
    while True:
        await asyncio.sleep(86400)  # 24 hours
        await cleanup_staging()

Verification Gate

Verification Gate

Run:

cd /home/david/Projects/keryx-bff && uv run python -c "
from src.services.staging import stage_file, validate_upload
valid = validate_upload(b'hello', 'text/plain')
assert valid == True
invalid_mime = validate_upload(b'hello', 'application/octet-stream')
assert invalid_mime == False
result = stage_file(b'test content', 'test.txt', 'text/plain')
assert result['name'] == 'test.txt'
assert result['size'] == 12
print('OK')"
# Expected output: OK

STOP Conditions

STOP Conditions

• If validate_upload accepts an unallowed MIME type, stop and fix.

Depends on: t-004

## Goal Implement the file upload staging service and the lifespan-scheduled cleanup task. ## References ## Reference - DESIGN.md §4 Config (staging_dir, max_upload_size_mb, staging_cleanup_days) - AGENT.md §10 Security Constraints (File upload safety: MIME validation, 50 MB cap) ## Implementation Steps ## Implementation Steps ### 1. Create `src/services/staging.py` (complete implementation) Replace placeholder with full module: ```python # src/services/staging.py import asyncio from pathlib import Path from loguru import logger import uuid from src.core.config import load_config def _allowed_mime_types() -> set[str]: return { "application/pdf", "text/plain", "text/html", "text/css", "text/javascript", "image/png", "image/jpeg", "image/gif", "image/webp", "application/json", "application/xml", "video/mp4", "audio/mpeg", "application/zip", } def validate_upload(file_data: bytes, mime_type: str) -> bool: config = load_config() max_bytes = config.bff.max_upload_size_mb * 1024 * 1024 if len(file_data) > max_bytes: return False if mime_type not in _allowed_mime_types(): return False return True def stage_file(file_data: bytes, name: str, mime_type: str) -> dict: config = load_config() staging_dir = Path(config.bff.staging_dir) staging_dir.mkdir(parents=True, exist_ok=True) file_id = uuid.uuid4().hex[:12] dest = staging_dir / f"{file_id}/{name}" dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(file_data) return { "staging_path": str(dest.relative_to(staging_dir)), "name": name, "size": len(file_data), "mime_type": mime_type, } async def cleanup_staging(): config = load_config() staging_dir = Path(config.bff.staging_dir) if not staging_dir.exists(): return cutoff_days = config.bff.staging_cleanup_days import time now = time.time() cutoff = now - (cutoff_days * 86400) for item in staging_dir.rglob("*"): if item.is_file() and item.stat().st_mtime < cutoff: item.unlink() logger.info(f"Cleaned up stale staging file: {item}") async def schedule_cleanup(): """Run cleanup every 24 hours.""" while True: await asyncio.sleep(86400) # 24 hours await cleanup_staging() ``` ## Verification Gate ## Verification Gate Run: ```bash cd /home/david/Projects/keryx-bff && uv run python -c " from src.services.staging import stage_file, validate_upload valid = validate_upload(b'hello', 'text/plain') assert valid == True invalid_mime = validate_upload(b'hello', 'application/octet-stream') assert invalid_mime == False result = stage_file(b'test content', 'test.txt', 'text/plain') assert result['name'] == 'test.txt' assert result['size'] == 12 print('OK')" # Expected output: OK ``` ## STOP Conditions ## STOP Conditions - If validate_upload accepts an unallowed MIME type, stop and fix. Depends on: t-004

Sign in to join this conversation.

No Branch/Tag specified

Branches Tags

main

No results found.

Labels

Clear labels

No items

No labels

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

1 participant

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference: david/keryx-bff#15

No description provided.