Pythonrestful_api.py
Copy
1
import hashlib
2
from fastapi import FastAPI, HTTPException, Response, Header
3
from pydantic import BaseModel
4
 
5
app = FastAPI()
6
 
7
class ArticleCreate(BaseModel):
8
    title: str
9
    content: str
10
 
11
class Article(ArticleCreate):
12
    id: int
13
    _links: dict = {}
14
 
15
articles_db: dict[int, Article] = {}
16
 
17
def make_links(article_id: int) -> dict:
18
    return {
19
        "self": {"href": f"/articles/{article_id}"},
20
        "collection": {"href": "/articles"},
21
    }
22
 
23
def etag(obj: dict) -> str:
24
    return hashlib.md5(str(obj).encode()).hexdigest()
25
 
26
@app.get("/articles")
27
async def list_articles():
28
    return {"articles": list(articles_db.values()), "_links": {"self": {"href": "/articles"}}}
29
 
30
@app.post("/articles", status_code=201)
31
async def create_article(article: ArticleCreate, response: Response):
32
    new_id = len(articles_db) + 1
33
    new_article = Article(id=new_id, **article.model_dump())
34
    articles_db[new_id] = new_article
35
    response.headers["Location"] = f"/articles/{new_id}"
36
    return {**new_article.model_dump(), "_links": make_links(new_id)}
37
 
38
@app.get("/articles/{article_id}")
39
async def get_article(article_id: int, if_none_match: str | None = Header(None)):
40
    if article_id not in articles_db:
41
        raise HTTPException(status_code=404, detail="Article not found")
42
    article = articles_db[article_id]
43
    tag = etag(article.model_dump())
44
    if if_none_match == tag:
45
        return Response(status_code=304)
46
    return Response(
47
        content=str({**article.model_dump(), "_links": make_links(article_id)}),
48
        headers={"ETag": tag},
49
        media_type="application/json",
50
    )
51
 
52
@app.put("/articles/{article_id}")
53
async def update_article(article_id: int, article: ArticleCreate):
54
    if article_id not in articles_db:
55
        raise HTTPException(status_code=404, detail="Article not found")
56
    updated = Article(id=article_id, **article.model_dump())
57
    articles_db[article_id] = updated
58
    return {**updated.model_dump(), "_links": make_links(article_id)}
59
 
60
@app.delete("/articles/{article_id}", status_code=204)
61
async def delete_article(article_id: int):
62
    if article_id not in articles_db:
63
        raise HTTPException(status_code=404, detail="Article not found")
64
    del articles_db[article_id]
65