-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
83 lines (64 loc) · 2.26 KB
/
main.py
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
from sqlite3 import IntegrityError
from slugify import UniqueSlugify
import databases
import datetime as dt
from fastapi import FastAPI, Request, Form, status
from fastapi.templating import Jinja2Templates
import sqlalchemy as sa
from fastapi.staticfiles import StaticFiles
from fastapi.responses import RedirectResponse
DATABASE_URL = "sqlite:///./blog.db"
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
database = databases.Database(DATABASE_URL)
metadata = sa.MetaData()
posts = sa.Table(
"posts",
metadata,
sa.Column("id", sa.Integer, primary_key=True),
sa.Column("title", sa.String, nullable=False),
sa.Column("subtitle", sa.String),
sa.Column("slug", sa.String, nullable=False, unique=True),
sa.Column("draft", sa.Boolean, nullable=False),
sa.Column("content", sa.Text, nullable=False),
sa.Column("publish_date", sa.DateTime, default=dt.datetime.utcnow),
)
engine = sa.create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
metadata.create_all(engine)
@app.get("/")
def index(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@app.get("/admin/post/")
def add_post_form(request: Request):
return templates.TemplateResponse("add_post.html", {"request": request})
@app.post("/admin/post/")
async def add_post(
request: Request,
title: str = Form(...),
subtitle: str = Form(...),
draft: bool = Form(False),
content: str = Form(...),
):
slugify = UniqueSlugify()
while True:
query = posts.insert().values(
title=title,
subtitle=subtitle,
draft=draft,
content=content,
slug=slugify(title, to_lower=True),
)
try:
await database.execute(query)
except IntegrityError:
continue
break
# TODO: Maybe redirect to newly created post
return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
@app.get("/about/")
def about(request: Request):
return templates.TemplateResponse("about.html", {"request": request})
@app.get("/contact/")
def contact(request: Request):
return templates.TemplateResponse("contact.html", {"request": request})