80 lines
2 KiB
Python
80 lines
2 KiB
Python
|
from fastapi import FastAPI, HTTPException
|
||
|
|
from pydantic import BaseModel
|
||
|
|
import psycopg2
|
||
|
|
import psycopg2.extras
|
||
|
|
import os
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
app = FastAPI(title="Notes API")
|
||
|
|
|
||
|
|
def get_db():
|
||
|
|
return psycopg2.connect(
|
||
|
|
host=os.environ["DB_HOST"],
|
||
|
|
port=int(os.environ.get("DB_PORT", "5432")),
|
||
|
|
database=os.environ["DB_NAME"],
|
||
|
|
user=os.environ["DB_USER"],
|
||
|
|
password=os.environ["DB_PASSWORD"],
|
||
|
|
)
|
||
|
|
|
||
|
|
@app.on_event("startup")
|
||
|
|
def startup():
|
||
|
|
conn = get_db()
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("""
|
||
|
|
CREATE TABLE IF NOT EXISTS notes (
|
||
|
|
id SERIAL PRIMARY KEY,
|
||
|
|
title VARCHAR(255) NOT NULL,
|
||
|
|
content TEXT DEFAULT '',
|
||
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
|
|
)
|
||
|
|
""")
|
||
|
|
conn.commit()
|
||
|
|
cur.close()
|
||
|
|
conn.close()
|
||
|
|
|
||
|
|
class NoteCreate(BaseModel):
|
||
|
|
title: str
|
||
|
|
content: Optional[str] = ""
|
||
|
|
|
||
|
|
@app.get("/health")
|
||
|
|
def health():
|
||
|
|
return {"status": "ok"}
|
||
|
|
|
||
|
|
@app.get("/notes")
|
||
|
|
def get_notes():
|
||
|
|
conn = get_db()
|
||
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||
|
|
cur.execute("SELECT * FROM notes ORDER BY created_at DESC")
|
||
|
|
notes = list(cur.fetchall())
|
||
|
|
cur.close()
|
||
|
|
conn.close()
|
||
|
|
return notes
|
||
|
|
|
||
|
|
@app.post("/notes")
|
||
|
|
def create_note(note: NoteCreate):
|
||
|
|
conn = get_db()
|
||
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||
|
|
cur.execute(
|
||
|
|
"INSERT INTO notes (title, content) VALUES (%s, %s) RETURNING *",
|
||
|
|
(note.title, note.content),
|
||
|
|
)
|
||
|
|
new_note = dict(cur.fetchone())
|
||
|
|
conn.commit()
|
||
|
|
cur.close()
|
||
|
|
conn.close()
|
||
|
|
return new_note
|
||
|
|
|
||
|
|
@app.delete("/notes/{note_id}")
|
||
|
|
def delete_note(note_id: int):
|
||
|
|
conn = get_db()
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("DELETE FROM notes WHERE id = %s", (note_id,))
|
||
|
|
if cur.rowcount == 0:
|
||
|
|
cur.close()
|
||
|
|
conn.close()
|
||
|
|
raise HTTPException(status_code=404, detail="Note not found")
|
||
|
|
conn.commit()
|
||
|
|
cur.close()
|
||
|
|
conn.close()
|
||
|
|
return {"deleted": note_id}
|