创建和搜索图书
首先让我们来添加创建、搜索图书的 API，然后再放入一些测试数据供后续使用。
创建图书 API
移动 domain/model 到 web/domain/model。
创建 web/domain/gateway/book_manager.py:
from abc import ABC, abstractmethod
from ..model import Book
class BookManager(ABC):
@abstractmethod
def create_book(self, b: Book) -> int:
pass
Python 中没有像其他编程语言那样有内置"接口"。
好在，你可以使用abc模块中的抽象基类（ABC）来实现类似的功能。
此处也是在使用4层架构。项目越大，该架构的好处越明显。阅读更多。
添加 web/domain/gateway/__init__.py:
from .book_manager import BookManager
安装 mysql 依赖：
pip3 install mysql-connector-python
该依赖是官方出品的 Python MySQL 驱动 。
更新 requirements.txt:
pip3 freeze > requirements.txt
准备 MySQL 数据库:
- 在你的机器上安装 MySQL 并启动。
- 创建一个库，名为
lr_event_book。CREATE DATABASE lr_event_book CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
- 创建用户
test_user:CREATE USER 'test_user'@'localhost' IDENTIFIED BY 'test_pass'; GRANT ALL PRIVILEGES ON lr_event_book.* TO 'test_user'@'localhost'; FLUSH PRIVILEGES;
创建 web/infrastructure/database/mysql.py:
import mysql.connector
from ...domain.gateway import BookManager
from ...domain.model import Book
from ..config import DBConfig
class MySQLPersistence(BookManager):
def __init__(self, c: DBConfig):
self.conn = mysql.connector.connect(
host=c.host,
port=c.port,
user=c.user,
password=c.password,
database=c.database,
autocommit=True
)
self.cursor = self.conn.cursor(dictionary=True)
self._create_table()
def _create_table(self):
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS books (
id INT AUTO_INCREMENT PRIMARY KEY,
title VARCHAR(255) NOT NULL,
author VARCHAR(255) NOT NULL,
published_at DATE NOT NULL,
description TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
''')
def create_book(self, b: Book) -> int:
self.cursor.execute('''
INSERT INTO books (title, author, published_at, description) VALUES (%s, %s, %s, %s)
''', (b.title, b.author, b.published_at, b.description))
return self.cursor.lastrowid or 0
添加 web/infrastructure/database/__init__.py:
from .mysql import MySQLPersistence
安装 yaml 依赖:
pip3 install pyyaml
更新 requirements.txt:
pip3 freeze > requirements.txt
创建 web/infrastructure/config/config.py:
from dataclasses import dataclass
import yaml
@dataclass
class DBConfig:
host: str
port: int
user: str
password: str
database: str
@dataclass
class ApplicationConfig:
port: int
page_size: int
templates_dir: str
@dataclass
class Config:
app: ApplicationConfig
db: DBConfig
def parseConfig(filename: str) -> Config:
with open(filename, 'r') as f:
data = yaml.safe_load(f)
return Config(
ApplicationConfig(**data['app']),
DBConfig(**data['db'])
)
添加 web/infrastructure/config/__init__.py:
from .config import Config, DBConfig, parseConfig
创建 web/config.yml:
app:
port: 8000
page_size: 5
templates_dir: "web/adapter/templates/"
db:
host: 127.0.0.1
port: 3306
user: "test_user"
password: "test_pass"
database: "lr_event_book"
警醒:
不要直接 git 提交 config.yml。可能会导致敏感数据泄露。如果非要提交的话，建议只提交配置格式模板。
比如：app: port: 8000 db: dsn: ""
创建 web/application/executor/book_operator.py:
from datetime import datetime
from .. import dto
from ...domain.model import Book
from ...domain.gateway import BookManager
class BookOperator():
def __init__(self, book_manager: BookManager):
self.book_manager = book_manager
def create_book(self, b: dto.Book) -> Book:
book = Book(id=0, created_at=datetime.now(), **b.__dict__)
id = self.book_manager.create_book(book)
book.id = id
return book
记得为每个子包添加 __init__.py。它用于辅助导出符号。
创建 web/application/wire_helper.py:
from ..domain.gateway import BookManager
from ..infrastructure.config import Config
from ..infrastructure.database import MySQLPersistence
class WireHelper:
def __init__(self, sqlPersistence: MySQLPersistence):
self.sqlPersistence = sqlPersistence
@classmethod
def new(cls, c: Config):
db = MySQLPersistence(c.db)
return cls(db)
def book_manager(self) -> BookManager:
return self.sqlPersistence
创建 web/application/dto/book.py:
from dataclasses import dataclass
@dataclass
class Book:
title: str
author: str
published_at: str
description: str
DTO 表示数据传输对象（Data Transfer Object）。 在应用程序中，数据的内部表示通常可能与你想通过 API 公开展示的表示方式不同。数据传输对象（DTO）允许你将内部数据结构转换为更适合 API 消费者的格式。
创建 web/application/__init__.py:
from .wire_helper import WireHelper
from . import dto
创建 web/adapter/router.py:
import logging
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from ..application.executor import BookOperator
from ..application import WireHelper, dto
from ..domain.model import Book
class RestHandler:
def __init__(self, logger: logging.Logger, book_operator: BookOperator):
self._logger = logger
self.book_operator = book_operator
def create_book(self, b: dto.Book):
try:
return self.book_operator.create_book(b)
except Exception as e:
self._logger.error(f"Failed to create: {e}")
raise HTTPException(status_code=400, detail="Failed to create")
def make_router(app: FastAPI, templates_dir: str, wire_helper: WireHelper):
rest_handler = RestHandler(
logging.getLogger("lr-event"),
BookOperator(wire_helper.book_manager())
)
templates = Jinja2Templates(directory=templates_dir)
@app.get("/", response_class=HTMLResponse)
async def index_page(request: Request):
return templates.TemplateResponse(
name="index.html", context={"request": request, "title": "LiteRank Book Store"}
)
@app.post("/api/books", response_model=Book)
async def create_book(b: dto.Book):
return rest_handler.create_book(b)
移动 templates 到 web/adapter/templates。
移动 main.py 到 web/main.py，并用如下代码替代其内容：
from fastapi import FastAPI
from .adapter.router import make_router
from .application import WireHelper
from .infrastructure.config import parseConfig
CONFIG_FILENAME = "web/config.yml"
c = parseConfig(CONFIG_FILENAME)
wire_helper = WireHelper.new(c)
app = FastAPI()
make_router(app, c.app.templates_dir, wire_helper)
再次运行 server，然后使用 curl 进行测试：
uvicorn web.main:app --reload
curl --request POST \
--url http://localhost:8000/api/books \
--header 'Content-Type: application/json' \
--data '{
"title": "The Great Gatsby",
"author": "F. Scott Fitzgerald",
"published_at": "1925-04-10",
"description": "A novel depicting the opulent lives of wealthy Long Island residents during the Jazz Age."
}'
样例响应：
{
"id": 1,
"title": "The Great Gatsby",
"author": "F. Scott Fitzgerald",
"published_at": "1925-04-10",
"description": "A novel depicting the opulent lives of wealthy Long Island residents during the Jazz Age.",
"created_at": "2024-04-10T01:20:52.505836"
}
放入测试数据
curl -X POST -H "Content-Type: application/json" -d '{"title": "To Kill a Mockingbird", "author": "Harper Lee", "published_at": "1960-07-11", "description": "A novel set in the American South during the 1930s, dealing with themes of racial injustice and moral growth."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "1984", "author": "George Orwell", "published_at": "1949-06-08", "description": "A dystopian novel depicting a totalitarian regime, surveillance, and propaganda."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "Pride and Prejudice", "author": "Jane Austen", "published_at": "1813-01-28", "description": "A classic novel exploring the themes of love, reputation, and social class in Georgian England."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "The Catcher in the Rye", "author": "J.D. Salinger", "published_at": "1951-07-16", "description": "A novel narrated by a disaffected teenager, exploring themes of alienation and identity."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "The Lord of the Rings", "author": "J.R.R. Tolkien", "published_at": "1954-07-29", "description": "A high fantasy epic following the quest to destroy the One Ring and defeat the Dark Lord Sauron."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "Moby-Dick", "author": "Herman Melville", "published_at": "1851-10-18", "description": "A novel exploring themes of obsession, revenge, and the nature of good and evil."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "The Hobbit", "author": "J.R.R. Tolkien", "published_at": "1937-09-21", "description": "A fantasy novel set in Middle-earth, following the adventure of Bilbo Baggins and the quest for treasure."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "The Adventures of Huckleberry Finn", "author": "Mark Twain", "published_at": "1884-12-10", "description": "A novel depicting the journey of a young boy and an escaped slave along the Mississippi River."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "War and Peace", "author": "Leo Tolstoy", "published_at": "1869-01-01", "description": "A novel depicting the Napoleonic era in Russia, exploring themes of love, war, and historical determinism."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "Alice’s Adventures in Wonderland", "author": "Lewis Carroll", "published_at": "1865-11-26", "description": "A children’s novel featuring a young girl named Alice who falls into a fantastical world populated by peculiar creatures."}' http://localhost:8000/api/books
curl -X POST -H "Content-Type: application/json" -d '{"title": "The Odyssey", "author": "Homer", "published_at": "8th Century BC", "description": "An ancient Greek epic poem attributed to Homer, detailing the journey of Odysseus after the Trojan War."}' http://localhost:8000/api/books
搜索图书 API
修改 web/domain/gateway/book_manager.py:
@@ -1,4 +1,5 @@
from abc import ABC, abstractmethod
+from typing import List
from ..model import Book
@@ -7,3 +8,7 @@ class BookManager(ABC):
@abstractmethod
def create_book(self, b: Book) -> int:
pass
+
+ @abstractmethod
+ def get_books(self, offset: int, keyword: str) -> List[Book]:
+ pass
更新 web/infrastructure/database/mysql.py:
@@ -1,3 +1,5 @@
+from typing import Any, List
+
import mysql.connector
from ...domain.gateway import BookManager
@@ -6,7 +8,8 @@ from ..config import DBConfig
class MySQLPersistence(BookManager):
- def __init__(self, c: DBConfig):
+ def __init__(self, c: DBConfig, page_size: int):
+ self.page_size = page_size
self.conn = mysql.connector.connect(
host=c.host,
port=c.port,
@@ -35,3 +38,16 @@ class MySQLPersistence(BookManager):
INSERT INTO books (title, author, published_at, description) VALUES (%s, %s, %s, %s)
''', (b.title, b.author, b.published_at, b.description))
return self.cursor.lastrowid or 0
+
+ def get_books(self, offset: int, keyword: str) -> List[Book]:
+ query = "SELECT * FROM books"
+ params: List[Any] = []
+ if keyword:
+ query += " WHERE title LIKE %s OR author LIKE %s OR description LIKE %s"
+ params = [f"%{keyword}%", f"%{keyword}%", f"%{keyword}%"]
+ query += " LIMIT %s, %s"
+ params.extend([offset, self.page_size])
+
+ self.cursor.execute(query, tuple(params))
+ results: List[Any] = self.cursor.fetchall()
+ return [Book(**result) for result in results]
更新 web/application/executor/book_operator.py:
@@ -1,4 +1,5 @@
from datetime import datetime
+from typing import List
from .. import dto
from ...domain.model import Book
@@ -15,3 +16,6 @@ class BookOperator():
id = self.book_manager.create_book(book)
book.id = id
return book
+
+ def get_books(self, offset: int, query: str) -> List[Book]:
+ return self.book_manager.get_books(offset, query)
更新 web/adapter/router.py:
@@ -21,6 +21,14 @@ class RestHandler:
self._logger.error(f"Failed to create: {e}")
raise HTTPException(status_code=400, detail="Failed to create")
+ def get_books(self, offset: int, query: str):
+ try:
+ books = self.book_operator.get_books(offset, query)
+ return books
+ except Exception as e:
+ self._logger.error(f"Failed to get books: {e}")
+ raise HTTPException(status_code=404, detail="Failed to get books")
+
def make_router(app: FastAPI, templates_dir: str, wire_helper: WireHelper):
rest_handler = RestHandler(
@@ -31,11 +39,21 @@ def make_router(app: FastAPI, templates_dir: str, wire_helper: WireHelper):
templates = Jinja2Templates(directory=templates_dir)
@app.get("/", response_class=HTMLResponse)
- async def index_page(request: Request):
+ async def index_page(request: Request, q: str = ""):
+ books = rest_handler.book_operator.get_books(0, q)
return templates.TemplateResponse(
- name="index.html", context={"request": request, "title": "LiteRank Book Store"}
+ name="index.html", context={
+ "request": request,
+ "title": "LiteRank Book Store",
+ "books": books,
+ "q": q,
+ }
)
@app.post("/api/books", response_model=Book)
async def create_book(b: dto.Book):
return rest_handler.create_book(b)
+
+ @app.get("/api/books")
+ async def get_books(o: int = 0, q: str = ""):
+ return rest_handler.get_books(o, q)
重启后再次使用 curl 测试：
curl --request GET --url 'http://localhost:8000/api/books?q=love'
样例响应：
[
{
"id": 4,
"title": "Pride and Prejudice",
"author": "Jane Austen",
"published_at": "1813-01-28",
"description": "A classic novel exploring the themes of love, reputation, and social class in Georgian England.",
"created_at": "2024-04-02T21:02:59.314+08:00"
},
{
"id": 10,
"title": "War and Peace",
"author": "Leo Tolstoy",
"published_at": "1869-01-01",
"description": "A novel depicting the Napoleonic era in Russia, exploring themes of love, war, and historical determinism.",
"created_at": "2024-04-02T21:02:59.42+08:00"
}
]
在首页上展示图书
如前文的修改：
@@ -31,11 +39,21 @@ def make_router(app: FastAPI, templates_dir: str, wire_helper: WireHelper):
templates = Jinja2Templates(directory=templates_dir)
@app.get("/", response_class=HTMLResponse)
- async def index_page(request: Request):
+ async def index_page(request: Request, q: str = ""):
+ books = rest_handler.book_operator.get_books(0, q)
return templates.TemplateResponse(
- name="index.html", context={"request": request, "title": "LiteRank Book Store"}
+ name="index.html", context={
+ "request": request,
+ "title": "LiteRank Book Store",
+ "books": books,
+ "q": q,
+ }
)
更新 web/adapter/templates/index.html:
@@ -10,12 +10,29 @@
</head>
<body class="bg-gray-100 p-2">
<div class="container mx-auto py-8">
- <h1 class="text-4xl font-bold">{{ title }}</h1>
+ <h1 class="text-4xl font-bold"><a href="/">{{ title }}</a></h1>
<!-- Search Bar Section -->
<div class="mb-8">
<h2 class="text-2xl font-bold mb-4 mt-6">Search</h2>
- <input type="text" placeholder="Search for books..." class="w-full px-4 py-2 rounded-md border-gray-300 focus:outline-none focus:border-blue-500">
+ <form class="flex">
+ <input type="text" name="q" value="{{q}}" placeholder="Search for books..." class="flex-grow px-4 py-2 rounded-l-md border-gray-300 focus:outline-none focus:border-blue-500">
+ <button type="submit" class="px-4 py-2 bg-blue-500 text-white rounded-r-md">Search</button>
+ </form>
+ </div>
+
+ <!-- Books Section -->
+ <div class="mb-8">
+ <h2 class="text-2xl font-bold mb-4">{%if q%}Keyword: “{{q}}“{%else%}Books{%endif%}</h2>
+ <div class="grid grid-cols-4 gap-2">
+ {% for book in books %}
+ <div class="bg-white p-4 rounded-md border-gray-300 shadow mt-2">
+ <div><b>{{book.title}}</b></div>
+ <div class="text-gray-500 text-sm">{{book.published_at}}</div>
+ <div class="italic text-sm">{{book.author}}</div>
+ </div>
+ {% endfor %}
+ </div>
</div>
<!-- Trends Section -->
重启 server 并刷新页面，你将看到图书列表。
在首页上搜索图书
如前文修改，web/adapter/router.py:
- async def index_page(request: Request):
+ async def index_page(request: Request, q: str = ""):
+ books = rest_handler.book_operator.get_books(0, q)
return templates.TemplateResponse(
name="index.html", context={
"request": request,
"title": "LiteRank Book Store",
"books": books,
+ "q": q,
}
)
如前文修改，web/dapter/templates/index.html:
@@ -15,12 +15,15 @@
<!-- Search Bar Section -->
<div class="mb-8">
<h2 class="text-2xl font-bold mb-4 mt-6">Search</h2>
- <input type="text" placeholder="Search for books..." class="w-full px-4 py-2 rounded-md border-gray-300 focus:outline-none focus:border-blue-500">
+ <form class="flex">
+ <input type="text" name="q" value="{{q}}" placeholder="Search for books..." class="flex-grow px-4 py-2 rounded-l-md border-gray-300 focus:outline-none focus:border-blue-500">
+ <button type="submit" class="px-4 py-2 bg-blue-500 text-white rounded-r-md">Search</button>
+ </form>
</div>
<!-- Books Section -->
<div class="mb-8">
- <h2 class="text-2xl font-bold mb-4">Books</h2>
+ <h2 class="text-2xl font-bold mb-4">{%if q%}Keyword: “{{q}}“{%else%}Books{%endif%}</h2>
<div class="grid grid-cols-4 gap-2">
{{range .books}}
<div class="bg-white p-4 rounded-md border-gray-300 shadow mt-2">
重启 server，刷新页面。
搜索结果如下所示：
