Spaces:
Runtime error
Runtime error
File size: 3,364 Bytes
a240da9 126a4c6 a240da9 |
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 |
import logging
import uuid
from fastapi import FastAPI
from base import EntriesResult, JobStatus, JobStatusResult, RequestInput
from db import TABLES, get_db_cursor
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
app = FastAPI()
# status
@app.get("/status/")
def status() -> str:
return "OK"
@app.post("/submit/")
def submit_job(input: RequestInput) -> str:
# submit a new job, poor man's job queue
_id = uuid.uuid4().hex
logger.info(f"Submitting job for (_id={_id[:8]})")
with get_db_cursor() as cursor:
# create a job
query = "INSERT INTO jobs (entry_id, status) VALUES (?, ?)"
cursor.execute(query, (_id, "pending"))
# create an entry
query = "INSERT INTO entries (id, author, source) VALUES (?, ?, ?)"
cursor.execute(query, (_id, input.author, input.content))
return f"Submitted job {_id}"
@app.get("/check_status/{_id}")
def check_status(_id: str) -> JobStatusResult:
with get_db_cursor() as cursor:
cursor.execute(
"SELECT status, last_updated FROM jobs WHERE entry_id = ?", (_id,)
)
result = cursor.fetchone()
if result is None:
return JobStatusResult(id=_id, status=JobStatus.not_found, last_updated=None)
status, last_updated = result
return JobStatusResult(id=_id, status=status, last_updated=last_updated)
@app.get("/recent/")
def recent() -> list[EntriesResult]:
with get_db_cursor() as cursor:
# get the last 10 entries, join summary and tag, where each tag is
# joined to a comma separated str
cursor.execute("""
SELECT e.id, e.author, s.summary, GROUP_CONCAT(t.tag, ","), e.created_at
FROM entries e
JOIN summaries s ON e.id = s.entry_id
JOIN tags t ON e.id = t.entry_id
GROUP BY e.id
ORDER BY e.created_at DESC
LIMIT 10
""")
results = cursor.fetchall()
entries = []
for _id, author, summary, tags, date in results:
entry = EntriesResult(
id=_id, author=author, summary=summary, tags=tags.split(","), date=date
)
entries.append(entry)
return entries
@app.get("/recent/{tag}")
def recent_tag(tag: str) -> list[EntriesResult]:
if not tag.startswith("#"):
tag = "#" + tag
# same as recent, but filter by tag
with get_db_cursor() as cursor:
cursor.execute(
"""
SELECT e.id, e.author, s.summary, GROUP_CONCAT(t.tag, ","), e.created_at
FROM entries e
JOIN summaries s ON e.id = s.entry_id
JOIN tags t ON e.id = t.entry_id
WHERE t.tag = ?
GROUP BY e.id
ORDER BY e.created_at DESC
LIMIT 10
""",
(tag,),
)
results = cursor.fetchall()
entries = []
for _id, author, summary, tags, date in results:
entry = EntriesResult(
id=_id, author=author, summary=summary, tags=tags.split(","), date=date
)
entries.append(entry)
return entries
@app.get("/clear/")
def clear() -> str:
# clear all tables
logger.warning("Clearing all tables")
with get_db_cursor() as cursor:
for table_name in TABLES:
cursor.execute(f"DELETE FROM {table_name}")
return "OK"
|