mirror of
https://github.com/cyberbread-ru/notepad-server.git
synced 2026-07-14 12:51:04 +03:00
initial vibecoded... thing
This commit is contained in:
+25
@@ -0,0 +1,25 @@
|
||||
import os
|
||||
import secrets
|
||||
from fastapi import Security, HTTPException, status
|
||||
from fastapi.security import APIKeyHeader
|
||||
|
||||
API_KEY_NAME = "X-API-Key"
|
||||
_header_scheme = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
|
||||
|
||||
|
||||
def _load_key() -> str:
|
||||
key = os.getenv("API_KEY")
|
||||
if not key:
|
||||
raise RuntimeError("API_KEY environment variable is not set")
|
||||
return key
|
||||
|
||||
|
||||
def verify_api_key(api_key: str = Security(_header_scheme)) -> str:
|
||||
expected = _load_key()
|
||||
if not api_key or not secrets.compare_digest(api_key, expected):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid or missing API key",
|
||||
headers={"WWW-Authenticate": "X-API-Key"},
|
||||
)
|
||||
return api_key
|
||||
@@ -0,0 +1,25 @@
|
||||
from pydantic import BaseModel, Field
|
||||
from datetime import datetime
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
class NoteBase(BaseModel):
|
||||
title: str
|
||||
content: str
|
||||
|
||||
|
||||
class NoteCreate(NoteBase):
|
||||
pass
|
||||
|
||||
|
||||
class NoteUpdate(NoteBase):
|
||||
pass
|
||||
|
||||
|
||||
class Note(NoteBase):
|
||||
id: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
@@ -0,0 +1,45 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from app.auth import verify_api_key
|
||||
from app.models import Note, NoteCreate, NoteUpdate
|
||||
from app.services import NoteService
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/notes",
|
||||
tags=["notes"],
|
||||
dependencies=[Depends(verify_api_key)],
|
||||
)
|
||||
_service = NoteService()
|
||||
|
||||
|
||||
@router.get("/", response_model=list[Note])
|
||||
def get_notes():
|
||||
return _service.list()
|
||||
|
||||
|
||||
@router.get("/{note_id}", response_model=Note)
|
||||
def get_note(note_id: str):
|
||||
note = _service.get(note_id)
|
||||
if not note:
|
||||
raise HTTPException(status_code=404, detail="Note not found")
|
||||
return note
|
||||
|
||||
|
||||
@router.post("/", response_model=Note, status_code=201)
|
||||
def create_note(payload: NoteCreate):
|
||||
return _service.create(payload.title, payload.content)
|
||||
|
||||
|
||||
@router.put("/{note_id}", response_model=Note)
|
||||
def update_note(note_id: str, payload: NoteUpdate):
|
||||
note = _service.update(note_id, payload.title, payload.content)
|
||||
if not note:
|
||||
raise HTTPException(status_code=404, detail="Note not found")
|
||||
return note
|
||||
|
||||
|
||||
@router.delete("/{note_id}", status_code=204)
|
||||
def delete_note(note_id: str):
|
||||
deleted = _service.remove(note_id)
|
||||
if not deleted:
|
||||
raise HTTPException(status_code=404, detail="Note not found")
|
||||
@@ -0,0 +1,24 @@
|
||||
from typing import Optional
|
||||
|
||||
from app.models import Note
|
||||
from app.store import NoteStore
|
||||
|
||||
|
||||
class NoteService:
|
||||
def __init__(self):
|
||||
self._store = NoteStore()
|
||||
|
||||
def list(self) -> list[Note]:
|
||||
return self._store.find_all()
|
||||
|
||||
def get(self, note_id: str) -> Optional[Note]:
|
||||
return self._store.find_by_id(note_id)
|
||||
|
||||
def create(self, title: str, content: str) -> Note:
|
||||
return self._store.insert(title, content)
|
||||
|
||||
def update(self, note_id: str, title: str, content: str) -> Optional[Note]:
|
||||
return self._store.update(note_id, title, content)
|
||||
|
||||
def remove(self, note_id: str) -> bool:
|
||||
return self._store.delete(note_id)
|
||||
@@ -0,0 +1,89 @@
|
||||
import sqlite3
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
from uuid import uuid4
|
||||
|
||||
from app.models import Note
|
||||
|
||||
DB_PATH = "notes.db"
|
||||
|
||||
|
||||
def init_db():
|
||||
with get_conn() as conn:
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS notes (
|
||||
id TEXT PRIMARY KEY,
|
||||
title TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
)
|
||||
""")
|
||||
|
||||
|
||||
@contextmanager
|
||||
def get_conn():
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
conn.row_factory = sqlite3.Row
|
||||
try:
|
||||
yield conn
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _row_to_note(row: sqlite3.Row) -> Note:
|
||||
return Note(
|
||||
id=row["id"],
|
||||
title=row["title"],
|
||||
content=row["content"],
|
||||
created_at=datetime.fromisoformat(row["created_at"]),
|
||||
updated_at=datetime.fromisoformat(row["updated_at"]),
|
||||
)
|
||||
|
||||
|
||||
class NoteStore:
|
||||
def find_all(self) -> list[Note]:
|
||||
with get_conn() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM notes ORDER BY updated_at DESC"
|
||||
).fetchall()
|
||||
return [_row_to_note(r) for r in rows]
|
||||
|
||||
def find_by_id(self, note_id: str) -> Optional[Note]:
|
||||
with get_conn() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM notes WHERE id = ?", (note_id,)
|
||||
).fetchone()
|
||||
return _row_to_note(row) if row else None
|
||||
|
||||
def insert(self, title: str, content: str) -> Note:
|
||||
now = datetime.utcnow().isoformat()
|
||||
note = Note(
|
||||
id=str(uuid4()),
|
||||
title=title,
|
||||
content=content,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
with get_conn() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO notes (id, title, content, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
|
||||
(note.id, note.title, note.content, note.created_at.isoformat(), note.updated_at.isoformat()),
|
||||
)
|
||||
return note
|
||||
|
||||
def update(self, note_id: str, title: str, content: str) -> Optional[Note]:
|
||||
now = datetime.utcnow().isoformat()
|
||||
with get_conn() as conn:
|
||||
conn.execute(
|
||||
"UPDATE notes SET title = ?, content = ?, updated_at = ? WHERE id = ?",
|
||||
(title, content, now, note_id),
|
||||
)
|
||||
return self.find_by_id(note_id)
|
||||
|
||||
def delete(self, note_id: str) -> bool:
|
||||
with get_conn() as conn:
|
||||
cursor = conn.execute("DELETE FROM notes WHERE id = ?", (note_id,))
|
||||
return cursor.rowcount > 0
|
||||
Reference in New Issue
Block a user