利用 FastAPI 改寫自己的"開發者部落格"

使用 FastAPI 來改寫部落格的記錄

爲什麼要用 FastAPI，主要是他的寫法比較簡單，不容易出錯

它對於 ASGI 的 async ( 非同步 )的效能，的確不是蓋的！

規劃

記錄一，說明一下根目錄

1. blog.py 這是主要的程式templates 底下存放的是前臺的界面檔案

/-
  |-- blog.py
  |-- app
       |-- models.py
  |-- static
       |-- css
       |-- js
       |-- fonts
       |-- images
  |-- templates
       |-- layouts
            |-- base.html
            |-- base_post.html
       |-- posts
            |-- index.html
            |-- post.html

先寫 requirments.txt

這些是一邊測試得到的 直接用 pip install -r requirments.txt 安裝必要套件

fastapi
uvicorn
jinja2
aiofiles
sqlalchemy
PyMySQL
pydantic
typing
fastapi-sqlalchemy
api_pagination

開始寫 blog.py, models.py

因爲剛開始，所以 sqlalchemy 的建立資料庫，還在摸索它如何實現

• fastapi_sqlalchemy，感覺是給 sqlalchemy 外加一層來實作api_pagination 的方式去做成分頁

經過一天測試，還是自己找的實作的方法，暫時就這樣就可以了

blog.py

這裏改用自己主機，請修改 mysql+pymysql:// 帶入的值

from fastapi import FastAPI
from fastapi_sqlalchemy import DBSessionMiddleware  # middleware helper
from fastapi_sqlalchemy import db  # an object to provide global access to a database session
from fastapi.staticfiles import StaticFiles
from typing import Optional

from urllib.parse import parse_qsl, quote
from datetime import datetime, date, timedelta, timezone

from starlette.requests import Request
from starlette.templating import Jinja2Templates

from api_pagination import Paginator

from app.models import Post, Category

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")
app.add_middleware(DBSessionMiddleware, db_url="mysql+pymysql://root@localhost/blog")

templates = Jinja2Templates(directory="templates")

@app.get("/")
async def main(request: Request, page: Optional[int] = None):
    if not page:
        page = 1
    count = db.session.query(Post, Category).filter(Post.category_id == Category.id).count()
    paginator = Paginator(total=count, items_per_page=9)
    page_info = paginator.get_page_info(page=page)
    offset = (page - 1) * 9
    limit = page * 9
    posts = db.session.query(Post, Category).filter(Post.category_id == Category.id).offset(offset).limit(limit).all()
    category = db.session.query(Category).all()
    return templates.TemplateResponse("posts/index.html", {"request": request,
                                                           "posts": posts,
                                                           "category": category,
                                                           "count": count,
                                                           "page_info": page_info,
                                                           "category_name": "/"
                                                           }
                                      )

@app.get("/pages/{category_name}")
async def pages_get(request: Request, category_name: str, page: Optional[int] = None):
    if not page:
        page = 1
    count = db.session.query(Post, Category).filter(Post.category_id==Category.id, Category.name==category_name).count()
    paginator = Paginator(total=count, items_per_page=9)
    page_info = paginator.get_page_info(page=page)
    offset = (page - 1) * 9
    limit = page * 9
    posts = db.session.query(Post, Category).filter(Post.category_id==Category.id, Category.name==category_name).offset(offset).limit(limit).all()
    category = db.session.query(Category).all()
    return templates.TemplateResponse("posts/index.html", {"request": request,
                                                           "posts": posts,
                                                           "category": category,
                                                           "page_info": page_info,
                                                           'category_name': '/pages/' + category_name
                                                           }
                                      )


@app.get("/post/{slug_name}")
async def post_get(request: Request, slug_name: str):
    slug_name = quote(slug_name)
    post = db.session.query(Post).filter_by(slug=slug_name, published=1).first()
    category = db.session.query(Category).filter_by(id=post.category_id).first()
    category_all = db.session.query(Category).all()
    # 更新 view
    views = post.views
    q = db.session.query(Post).filter_by(id=post.id).first()
    q.views = views + 1
    db.session.commit()
    # 載入日期
    timestamp = datetime.now().timestamp()
    return templates.TemplateResponse("posts/post.html", {"request": request,
                                                           "post": post,
                                                           "category": category,
                                                           "category_all": category_all,
                                                           "timestamp": timestamp}
                                      )

models.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Date, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime, date, timedelta, timezone

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True)
    hashed_password = Column(String(255))
    is_active = Column(Boolean(), default=True)

class Items(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), index=True)
    description = Column(String(255), index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    category_id = Column(Integer)
    title = Column(String(255))
    content = Column(String)
    short_content = Column(String(255))
    slug = Column(String(255))
    published = Column(Integer)
    views = Column(Integer)
    image = Column(String(255))
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)  

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

資料庫 schema

-- ----------------------------
-- Table structure for categories
-- ----------------------------
DROP TABLE IF EXISTS `categories`;
CREATE TABLE `categories`  (
  `id` int(10) UNSIGNED NOT NULL AUTO_INCREMENT,
  `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,
  `created_at` timestamp(0) NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  `updated_at` timestamp(0) NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 9 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci ROW_FORMAT = Dynamic;

-- ----------------------------
-- Table structure for posts
-- ----------------------------
DROP TABLE IF EXISTS `posts`;
CREATE TABLE `posts`  (
  `id` int(11) UNSIGNED NOT NULL AUTO_INCREMENT,
  `user_id` int(11) NULL DEFAULT NULL,
  `category_id` int(11) NULL DEFAULT 1,
  `title` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,
  `slug` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,
  `views` int(11) NULL DEFAULT 0,
  `image` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,
  `content` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,
  `short_content` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,
  `published` int(1) NULL DEFAULT 1,
  `created_at` timestamp(0) NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  `updated_at` timestamp(0) NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 13 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci ROW_FORMAT = Dynamic;

後面有自己的 templates 的樣板，裏面稍微動了一下，改變幅度不大

自己的部落格就修改好了