Spaces:
Paused
Paused
File size: 3,695 Bytes
d044ca3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | import sqlite3
from datetime import datetime
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
BASE_DIR = Path(__file__).resolve().parent
DB_PATH = BASE_DIR / "notes.db"
app = FastAPI(title="Quick Notes MVP")
class NotePayload(BaseModel):
content: str = Field(..., min_length=1)
def get_connection() -> sqlite3.Connection:
connection = sqlite3.connect(DB_PATH)
connection.row_factory = sqlite3.Row
return connection
def init_db() -> None:
connection = get_connection()
connection.execute(
"""
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
connection.commit()
connection.close()
def row_to_dict(row: sqlite3.Row) -> dict:
return {
"id": row["id"],
"content": row["content"],
"created_at": row["created_at"],
"updated_at": row["updated_at"],
}
@app.on_event("startup")
def on_startup() -> None:
init_db()
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
@app.get("/")
def home() -> FileResponse:
return FileResponse(BASE_DIR / "static" / "index.html")
@app.get("/notes")
def get_notes() -> list[dict]:
connection = get_connection()
rows = connection.execute(
(
"SELECT id, content, created_at, updated_at "
"FROM notes ORDER BY id DESC"
)
).fetchall()
connection.close()
return [row_to_dict(row) for row in rows]
@app.post("/notes")
def create_note(payload: NotePayload) -> dict:
content = payload.content.strip()
if not content:
raise HTTPException(
status_code=400,
detail="Note content cannot be empty",
)
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
connection = get_connection()
cursor = connection.execute(
"INSERT INTO notes (content, created_at, updated_at) VALUES (?, ?, ?)",
(content, now, now),
)
connection.commit()
note_id = cursor.lastrowid
row = connection.execute(
"SELECT id, content, created_at, updated_at FROM notes WHERE id = ?",
(note_id,),
).fetchone()
connection.close()
return row_to_dict(row)
@app.put("/notes/{note_id}")
def update_note(note_id: int, payload: NotePayload) -> dict:
content = payload.content.strip()
if not content:
raise HTTPException(
status_code=400,
detail="Note content cannot be empty",
)
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
connection = get_connection()
cursor = connection.execute(
"UPDATE notes SET content = ?, updated_at = ? WHERE id = ?",
(content, now, note_id),
)
connection.commit()
if cursor.rowcount == 0:
connection.close()
raise HTTPException(status_code=404, detail="Note not found")
row = connection.execute(
"SELECT id, content, created_at, updated_at FROM notes WHERE id = ?",
(note_id,),
).fetchone()
connection.close()
return row_to_dict(row)
@app.delete("/notes/{note_id}")
def delete_note(note_id: int) -> dict:
connection = get_connection()
cursor = connection.execute("DELETE FROM notes WHERE id = ?", (note_id,))
connection.commit()
connection.close()
if cursor.rowcount == 0:
raise HTTPException(status_code=404, detail="Note not found")
return {"message": "Note deleted"}
|