from typing import Union
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
import uvicorn
import sqlite3
from contextlib import closing
from random import choice
from pydantic import BaseModel
app = FastAPI()
con = sqlite3.connect("file:quotes.sqlite3?mode=ro", uri=True, check_same_thread=False)
con1 = sqlite3.connect("file:tmp1?mode=memory&cache=shared", uri=True)
class Quote(BaseModel):
id: int
text: str
datetime: int
@app.get("/")
async def read_root():
return {"Hello": "World"}
def htmlize(data: str):
text = data.replace('\n', '
').strip()
container = f'Quote{text}'
return container
def to_href(id: int):
return f"{id} "
@app.get("/items/all")
async def get_all(format: Union[str, None] = None):
with closing(con.cursor()) as cursor:
data = cursor.execute("SELECT \"id\" from `quotes`").fetchall()
#id_list = []
#for i in data:
# id_list.append(to_href(i[0]))
if format=='text':
id_list = [str(i[0]) for i in data]
text = '\n'.join(id_list)
return PlainTextResponse(content=text)
else:
id_list = [to_href(i[0]) for i in data]
text = ''.join(id_list)
return HTMLResponse(content=htmlize(text), status_code=200)
#return None
@app.get("/items/random")
async def read_random():
with closing(con.cursor()) as cursor:
data = cursor.execute("SELECT \"id\" from `quotes`").fetchall()
choice_target = choice(data)
text = cursor.execute("SELECT * from `quotes` WHERE id=?", choice_target).fetchone()
return PlainTextResponse(content=text[1], status_code=200)
@app.get("/items/getrandom")
async def read_getrandom():
with closing(con.cursor()) as cursor:
data = cursor.execute("SELECT \"id\" from `quotes`").fetchall()
id_list = [i[0] for i in data]
to_id = choice(id_list)
print(to_id)
return RedirectResponse(url=f'/items/{to_id}')
#return None
@app.get("/random")
async def redirect_to_random():
return RedirectResponse(url="/items/getrandom")
@app.get("/items/{item_id}")
async def read_item(item_id: int, format: Union[str, None] = None):
with closing(con.cursor()) as cursor:
text = cursor.execute("SELECT * from `quotes` WHERE id=?", (item_id,)).fetchone()
print(text)
if text:
if format == 'json':
return {"id:": text[0], "text": text[1], "time:": text[2]}
elif format == 'html':
return HTMLResponse(content=htmlize(text[1]), status_code=200)
elif format == 'raw':
return text
elif format == 'rawtext':
return text[1]
else:
return PlainTextResponse(content=text[1], status_code=200)
else:
#return HTMLResponse(content=None, status_code=404)
raise HTTPException(status_code=404, detail="Item not found")
return None
@app.get("/items/{item_id}/next")
async def redirect_to_next(item_id: int):
return RedirectResponse(url=f"/items/{item_id+1}")
@app.get("/items/{item_id}/prev")
async def redirect_to_previous(item_id: int):
return RedirectResponse(url=f"/items/{item_id-1}")
if __name__ == "__main__":
uvicorn.run(app, port=8001, host="0.0.0.0")