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
| import logging
import traceback
from fastapi import APIRouter, Depends, HTTPException, Response, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from fastapi_sqlalchemy_v1.book import Book
from fastapi_sqlalchemy_v1.book_repository import (
BookAddParameter,
BookNotFoundError,
BookRepository,
)
from fastapi_sqlalchemy_v1.config import AppConfig
from fastapi_sqlalchemy_v1.database import Database
logger = logging.getLogger(__name__)
router = APIRouter()
class BookResponseHTTPEntity(BaseModel):
id: int = Field(...)
name: str = Field(..., example="book")
lang2: str = Field(..., example="ja")
created_at: str = Field(...)
@classmethod
def new(self, book: Book) -> "BookResponseHTTPEntity":
return BookResponseHTTPEntity(
id=book.book_id,
name=book.name,
lang2=book.lang2,
created_at=book.created_at.isoformat(),
)
class BookAddParameterHTTPEntity(BaseModel):
name: str = Field(..., min_length=1, max_length=20)
lang2: str = Field(..., min_lengh=2, max_length=2, example="ja")
def book_repository(session: Session) -> BookRepository:
return BookRepository(session)
@router.get("/book/{book_id}", response_model=BookResponseHTTPEntity, tags=["Book"])
def get_book(book_id: int, session: Session = Depends(Database.get_session)):
logger.info("get_book")
try:
book_repo: BookRepository = book_repository(session)
book: Book = book_repo.find_book_by_id(book_id)
book_http_entity = BookResponseHTTPEntity.new(book)
return book_http_entity
except BookNotFoundError:
raise HTTPException(status_code=404, detail="item_not_found")
except Exception as e:
logger.error(str(e))
logger.error(traceback.format_exc())
raise HTTPException(status_code=500, detail="internal server error")
@router.post("/book", tags=["Book"], responses={201: {"model": None}})
def add_book(
body: BookAddParameterHTTPEntity, session: Session = Depends(Database.get_session)
):
logger.info("add_book")
app_config = AppConfig.config.app
try:
book_add_param = BookAddParameter(name=body.name, lang2=body.lang2)
book_repo: BookRepository = book_repository(session)
book_id = book_repo.add_book(book_add_param)
url = f"{app_config.scheme}://{app_config.host}/book/{book_id}"
return Response(status_code=status.HTTP_201_CREATED, headers={"location": url})
except Exception as e:
logger.error(str(e))
logger.error(traceback.format_exc())
raise HTTPException(status_code=500, detail="internal server error")
|