Paste create api

modules/sqlite_helpers.py

@@ -23,12 +23,19 @@ def maybe_create_table(sqlite_file: str) -> bool:
             used INTEGER DEFAULT 1,
             expires_at DATETIME DEFAULT NULL);
         """
-
+        create_pastes_table_query = """
+        CREATE TABLE IF NOT EXISTS pastes (
+            paste_id TEXT PRIMARY KEY,
+            title TEXT,
+            created_at DATETIME DEFAULT CURRENT_DATE,
+            deleted_at DATETIME DEFAULT NULL,
+            expires_at DATETIME DEFAULT NULL);
+        """
         create_index_query = """
         CREATE UNIQUE INDEX IF NOT EXISTS idx_urls_alias
         ON urls (alias);
         """
-
+        cursor.execute(create_pastes_table_query)
         cursor.execute(create_table_query)
         cursor.execute(create_index_query)
         db.commit()
@@ -37,6 +44,24 @@ def maybe_create_table(sqlite_file: str) -> bool:
         logger.exception("Unable to create urls table")
         return False
 
+def insert_paste(sqlite_file: str, paste_id: str, title: str):
+    db = sqlite3.connect(sqlite_file)
+    cursor = db.cursor()
+
+    try:
+        sql = "INSERT INTO pastes(paste_id, title) VALUES (?, ?)"
+        val = (paste_id, title)
+        cursor.execute(sql, val)
+        db.commit()
+        return True
+    except sqlite3.IntegrityError:
+        return False
+    except Exception:
+        logger.exception("Inserting paste had an error")
+        return False
+    finally:
+        cursor.close()
+        db.close()
 
 def insert_url(sqlite_file: str, url: str, alias: str, expiration_date: typing.Union[str, None] = None):
     db = sqlite3.connect(sqlite_file)

requirements.txt

@@ -5,4 +5,6 @@ mysql-connector-python==8.0.33
 py-grpc-prometheus==0.7.0
 pyqrcode==1.2.1
 pypng==0.20220715.0
-pillow==10.2.0
+pillow==10.2.0
+python-multipart==0.0.9
+httpx==0.24.1

server.py

@@ -1,5 +1,5 @@
 from typing import Optional
-from fastapi import FastAPI, Request, HTTPException, Response
+from fastapi import FastAPI, Request, HTTPException, Response, File, UploadFile
 from fastapi.responses import RedirectResponse, HTMLResponse, FileResponse
 from fastapi.middleware.cors import CORSMiddleware
 import logging
@@ -17,6 +17,8 @@
 from modules.sqlite_helpers import increment_used_column
 from modules.cache import Cache
 from modules.qr_code import QRCode
+from pathlib import Path
+import secrets
 
 
 app = FastAPI()
@@ -34,6 +36,8 @@
 
 # maybe create the table if it doesnt already exist
 DATABASE_FILE = args.database_file_path
+MAX_PASTE_SIZE_BYTES = 10 * 1024 * 1024
+PASTES_DIR = Path("pastes")
 sqlite_helpers.maybe_create_table(DATABASE_FILE)
 qr_code_cache = QRCode(
   base_url=args.qr_code_base_url,
@@ -91,6 +95,33 @@ async def create_url(request: Request):
         logging.exception(f'returning 422 due to invalid alias of "{alias}"')
         raise HTTPException(status_code=HttpResponse.INVALID_ARGUMENT_EXCEPTION.code)
 
+@app.post("/paste/create")
+async def create_paste(file: UploadFile = File(...)):
+    content = await file.read(MAX_PASTE_SIZE_BYTES + 1)
+
+    if len(content) > MAX_PASTE_SIZE_BYTES:
+        raise HTTPException(status_code=400, detail="Paste file too large")
+
+    paste_id = secrets.token_hex(3)
+    PASTES_DIR.mkdir(exist_ok=True)
+
+    paste_path = PASTES_DIR / paste_id
+    paste_path.write_bytes(content)
+
+    inserted = sqlite_helpers.insert_paste(
+        DATABASE_FILE,
+        paste_id,
+        file.filename
+    )
+
+    if not inserted:
+        paste_path.unlink(missing_ok=True)
+        raise HTTPException(status_code=500, detail="Failed to create paste")
+
+    return {
+        "paste_id": paste_id,
+        "filename": file.filename
+    }
 
 @app.get("/list")
 async def get_urls(

test/test_pastes.py

@@ -0,0 +1,95 @@
+import os
+import sqlite3
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+# this allows imports from the modules folder to work
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from fastapi.testclient import TestClient
+from modules import sqlite_helpers
+
+# server.py requires command-line args when imported.
+# These fake args let the test import server.py without crashing.
+TEST_ROOT_DIR = tempfile.TemporaryDirectory()
+TEST_DB_PATH = os.path.join(TEST_ROOT_DIR.name, "test.db")
+TEST_QR_CACHE_DIR = tempfile.TemporaryDirectory()
+
+with mock.patch.object(
+    sys,
+    "argv",
+    [
+        "server.py",
+        "--database-file-path",
+        TEST_DB_PATH,
+        "--qr-code-cache-path",
+        TEST_QR_CACHE_DIR.name,
+        "--qr-code-base-url",
+        "http://localhost:8000",
+    ],
+):
+    import server
+
+
+class TestPasteEndpoints(unittest.TestCase):
+    def test_create_paste_rejects_file_larger_than_10mb(self):
+        large_content = b"a" * ((10 * 1024 * 1024) + 1)
+
+        with TestClient(server.app) as client:
+            response = client.post(
+                "/paste/create",
+                files={"file": ("large.txt", large_content, "text/plain")},
+            )
+
+        self.assertEqual(response.status_code, 400)
+
+    def test_create_paste_creates_file_and_database_row(self):
+        with tempfile.TemporaryDirectory() as tmp_root:
+            tmp_db_path = os.path.join(tmp_root, "test.db")
+            tmp_pastes_dir = os.path.join(tmp_root, "pastes")
+
+            sqlite_helpers.maybe_create_table(tmp_db_path)
+
+            with mock.patch.object(server, "DATABASE_FILE", tmp_db_path):
+                with mock.patch.object(server, "PASTES_DIR", Path(tmp_pastes_dir)):
+                    with mock.patch("server.secrets.token_hex", return_value="fe80df"):
+                        with TestClient(server.app) as client:
+                            response = client.post(
+                                "/paste/create",
+                                files={
+                                    "file": (
+                                        "example.txt",
+                                        b"hello paste",
+                                        "text/plain",
+                                    )
+                                },
+                            )
+
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json()["paste_id"], "fe80df")
+            self.assertEqual(response.json()["filename"], "example.txt")
+
+            paste_path = Path(tmp_pastes_dir) / "fe80df"
+            self.assertTrue(paste_path.exists())
+            self.assertEqual(paste_path.read_bytes(), b"hello paste")
+
+            db = sqlite3.connect(tmp_db_path)
+            cursor = db.cursor()
+            cursor.execute(
+                "SELECT paste_id, title FROM pastes WHERE paste_id = ?",
+                ("fe80df",),
+            )
+            row = cursor.fetchone()
+            cursor.close()
+            db.close()
+
+            self.assertIsNotNone(row)
+            self.assertEqual(row[0], "fe80df")
+            self.assertEqual(row[1], "example.txt")
+
+
+if __name__ == "__main__":
+    unittest.main()