mirror of
https://github.com/cyberbread-ru/notepad-server.git
synced 2026-07-14 12:51:04 +03:00
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
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
|