import sqlite3
from fastapi import FastAPI
from fastapi import Request, Form
from fastapi.responses import HTMLResponse, RedirectResponse
DB_PATH = "./data/database.db"
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, login TEXT NOT NULL UNIQUE, password TEXT NOT NULL)")
app = FastAPI(
docs_url=None, # Disable Swagger UI
redoc_url=None, # Disable ReDoc
openapi_url=None # Disable OpenAPI JSON schema
)
STYLES = """
"""
@app.get("/register", response_class=HTMLResponse)
async def register_form():
return f"""
Регистрация{STYLES}
Регистрация
Уже есть аккаунт? Войти
"""
@app.post("/register")
async def register(login: str = Form(...), password: str = Form(...)):
try:
cursor.execute("INSERT INTO users (login, password) VALUES (?, ?)", (login, password))
conn.commit()
response = RedirectResponse(url="/welcome", status_code=302)
response.set_cookie("login", login)
response.set_cookie("password", password)
return response
except sqlite3.IntegrityError:
return HTMLResponse(f"""
Ошибка регистрации{STYLES}
Login уже существует
Попробовать снова
""", status_code=400)
@app.get("/login", response_class=HTMLResponse)
async def login_form():
return f"""
Вход{STYLES}
Вход
Нет аккаунта? Зарегистрироваться
"""
@app.post("/login")
async def login(login: str = Form(...), password: str = Form(...)):
cursor.execute("SELECT * FROM users WHERE login=? AND password=?", (login, password))
user = cursor.fetchone()
if user:
response = RedirectResponse(url="/welcome", status_code=302)
response.set_cookie("login", login)
response.set_cookie("password", password)
return response
else:
return HTMLResponse(f"""
Ошибка входа{STYLES}
Неверные учетные данные
Попробовать снова
""", status_code=401)
@app.get("/welcome", response_class=HTMLResponse)
async def welcome(request: Request):
login = request.cookies.get("login")
password = request.cookies.get("password")
if not login or not password:
return RedirectResponse(url="/login")
cursor.execute("SELECT * FROM users WHERE login=? AND password=?", (login, password))
user = cursor.fetchone()
if user:
return f"""
Добро пожаловать{STYLES}
Привет, {login}
"""
else:
return RedirectResponse(url="/login")
@app.post("/logout")
async def logout():
response = RedirectResponse(url="/login", status_code=302)
response.delete_cookie("login")
response.delete_cookie("password")
return response
@app.get("/", include_in_schema=False)
async def root():
return RedirectResponse(url="/login")