diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/api/public.py b/backend/app/api/public.py new file mode 100644 index 0000000..9657ba8 --- /dev/null +++ b/backend/app/api/public.py @@ -0,0 +1,50 @@ +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from sqlalchemy.orm import Session + +from app.db import get_db +from app.schemas import ItemCreate, ItemOut, WallResponse +from app.services import items as items_service + +router = APIRouter(prefix="/api/items", tags=["items"]) + + +def _client_ip(request: Request) -> str: + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + return forwarded.split(",")[0].strip() + if request.client and request.client.host: + return request.client.host + return "" + + +@router.post("", response_model=ItemOut) +def create_item( + payload: ItemCreate, + request: Request, + db: Session = Depends(get_db), +): + item = items_service.create_text_item(db, payload, _client_ip(request)) + return items_service.item_to_dict(item) + + +@router.get("/wall", response_model=WallResponse) +def wall( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + rows, total = items_service.list_wall(db, page=page, page_size=page_size) + return WallResponse( + items=[ItemOut.model_validate(items_service.item_to_dict(i)) for i in rows], + page=page, + page_size=page_size, + total=total, + ) + + +@router.get("/{slug}", response_model=ItemOut) +def get_item(slug: str, db: Session = Depends(get_db)): + item = items_service.get_item(db, slug) + if item is None: + raise HTTPException(status_code=404, detail="Not found") + return items_service.item_to_dict(item) diff --git a/backend/app/main.py b/backend/app/main.py index ab21d0d..e38dede 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2,6 +2,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI +from app.api.public import router as public_router from app.config import settings from app.db import Base, engine from app import models # noqa: F401 — register models with Base.metadata @@ -16,6 +17,7 @@ async def lifespan(_app: FastAPI): app = FastAPI(title="MincedPad", lifespan=lifespan) +app.include_router(public_router) @app.get("/api/health") diff --git a/backend/app/schemas.py b/backend/app/schemas.py new file mode 100644 index 0000000..b47def3 --- /dev/null +++ b/backend/app/schemas.py @@ -0,0 +1,39 @@ +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, Field + + +TTLOption = Literal["1h", "24h", "7d", "never"] + + +class ItemCreate(BaseModel): + body: str = Field(min_length=1) + title: str | None = None + is_public: bool = True + burn_after_read: bool = False + ttl: TTLOption = "24h" + + +class ItemOut(BaseModel): + slug: str + kind: str + title: str + body: str | None = None + file_name: str | None = None + mime: str | None = None + size_bytes: int | None = None + is_public: bool + burn_after_read: bool + expires_at: datetime | None + created_at: datetime + view_count: int + + model_config = {"from_attributes": True} + + +class WallResponse(BaseModel): + items: list[ItemOut] + page: int + page_size: int + total: int diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/services/items.py b/backend/app/services/items.py new file mode 100644 index 0000000..084d879 --- /dev/null +++ b/backend/app/services/items.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import secrets +from datetime import datetime, timedelta +from pathlib import Path + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.config import settings +from app.models import Item +from app.schemas import ItemCreate, TTLOption + +TTL_DELTAS: dict[TTLOption, timedelta | None] = { + "1h": timedelta(hours=1), + "24h": timedelta(days=1), + "7d": timedelta(days=7), + "never": None, +} + + +def _slug() -> str: + return secrets.token_urlsafe(8) + + +def _expires_at(ttl: TTLOption, now: datetime | None = None) -> datetime | None: + delta = TTL_DELTAS[ttl] + if delta is None: + return None + return (now or datetime.utcnow()) + delta + + +def _title_from_body(body: str) -> str: + first = body.split("\n", 1)[0].strip() + return first[:255] + + +def _is_unavailable(item: Item, now: datetime | None = None) -> bool: + if item.burned: + return True + now = now or datetime.utcnow() + if item.expires_at is not None and item.expires_at <= now: + return True + return False + + +def item_to_dict(item: Item, *, include_body: bool = True) -> dict: + data = { + "slug": item.slug, + "kind": item.kind, + "title": item.title, + "file_name": item.file_name, + "mime": item.mime, + "size_bytes": item.size_bytes, + "is_public": item.is_public, + "burn_after_read": item.burn_after_read, + "expires_at": item.expires_at, + "created_at": item.created_at, + "view_count": item.view_count, + } + if include_body: + data["body"] = item.body + return data + + +def create_text_item(db: Session, payload: ItemCreate, created_ip: str) -> Item: + now = datetime.utcnow() + title = payload.title if payload.title is not None else _title_from_body(payload.body) + item = Item( + slug=_slug(), + kind="text", + title=title[:255], + body=payload.body, + is_public=payload.is_public, + burn_after_read=payload.burn_after_read, + expires_at=_expires_at(payload.ttl, now), + created_ip=created_ip, + created_at=now, + ) + db.add(item) + db.commit() + db.refresh(item) + return item + + +def get_item(db: Session, slug: str) -> Item | None: + item = db.scalar(select(Item).where(Item.slug == slug)) + if item is None or _is_unavailable(item): + return None + + item.view_count += 1 + if item.burn_after_read: + item.burned = True + if item.file_path: + path = Path(item.file_path) + if not path.is_absolute(): + path = settings.data_dir / path + try: + path.unlink(missing_ok=True) + except OSError: + pass + db.commit() + db.refresh(item) + return item + + +def list_wall( + db: Session, + *, + page: int = 1, + page_size: int = 20, +) -> tuple[list[Item], int]: + page = max(page, 1) + page_size = min(max(page_size, 1), 100) + now = datetime.utcnow() + + filters = [ + Item.is_public.is_(True), + Item.burned.is_(False), + (Item.expires_at.is_(None) | (Item.expires_at > now)), + ] + total = db.scalar(select(func.count()).select_from(Item).where(*filters)) or 0 + items = list( + db.scalars( + select(Item) + .where(*filters) + .order_by(Item.created_at.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + ).all() + ) + return items, total diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..13b4b35 --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,31 @@ +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + + +@pytest.fixture() +def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + + from app.config import Settings + import app.config as config_module + import app.db as db_module + import app.main as main_module + + settings = Settings(_env_file=None, DATA_DIR=tmp_path) + config_module.settings = settings + + engine = create_engine( + settings.db_url, + connect_args={"check_same_thread": False}, + ) + SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) + db_module.engine = engine + db_module.SessionLocal = SessionLocal + main_module.engine = engine + + with TestClient(main_module.app) as test_client: + yield test_client diff --git a/backend/tests/test_items.py b/backend/tests/test_items.py new file mode 100644 index 0000000..cb606c9 --- /dev/null +++ b/backend/tests/test_items.py @@ -0,0 +1,70 @@ +from datetime import datetime, timedelta + + +def test_create_text_default_public_and_24h(client): + r = client.post("/api/items", json={"body": "hello **md**"}) + assert r.status_code == 200 + data = r.json() + assert data["slug"] + assert data["is_public"] is True + assert data["expires_at"] is not None + assert data["kind"] == "text" + assert data["title"] == "hello **md**" + assert data["body"] == "hello **md**" + assert data["burn_after_read"] is False + assert data["view_count"] == 0 + assert data["created_at"] + + +def test_link_only_not_on_wall(client): + r = client.post("/api/items", json={"body": "secret", "is_public": False}) + slug = r.json()["slug"] + wall = client.get("/api/items/wall").json()["items"] + assert all(i["slug"] != slug for i in wall) + assert client.get(f"/api/items/{slug}").status_code == 200 + + +def test_ttl_options(client): + never = client.post("/api/items", json={"body": "n", "ttl": "never"}).json() + assert never["expires_at"] is None + + one_h = client.post("/api/items", json={"body": "h", "ttl": "1h"}).json() + expires = datetime.fromisoformat(one_h["expires_at"]) + delta = expires - datetime.fromisoformat(one_h["created_at"]) + assert timedelta(minutes=50) < delta < timedelta(hours=2) + + week = client.post("/api/items", json={"body": "w", "ttl": "7d"}).json() + expires_w = datetime.fromisoformat(week["expires_at"]) + delta_w = expires_w - datetime.fromisoformat(week["created_at"]) + assert timedelta(days=6) < delta_w < timedelta(days=8) + + +def test_wall_pagination_shape(client): + for i in range(3): + client.post("/api/items", json={"body": f"item {i}", "is_public": True}) + + r = client.get("/api/items/wall", params={"page": 1, "page_size": 2}) + assert r.status_code == 200 + data = r.json() + assert "items" in data + assert data["page"] == 1 + assert data["page_size"] == 2 + assert data["total"] >= 3 + assert len(data["items"]) == 2 + # newest first + titles = [i["title"] for i in data["items"]] + assert titles[0] == "item 2" + + +def test_get_missing_404(client): + assert client.get("/api/items/does-not-exist").status_code == 404 + + +def test_title_from_first_line(client): + r = client.post("/api/items", json={"body": "First line\nSecond line"}) + assert r.json()["title"] == "First line" + + +def test_explicit_title(client): + r = client.post("/api/items", json={"body": "body text", "title": "Custom"}) + assert r.json()["title"] == "Custom"