bookstore-api/routes.py at master · PskProduction/bookstore-api · GitHub

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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
from typing import Union

from flasgger import Swagger
from flask import Flask, request
from flask_restful import Api, Resource
from marshmallow import ValidationError
from werkzeug.serving import WSGIRequestHandler

from models import (
    DATA,
    get_all_books,
    init_db,
    add_book,
    get_book_by_id,
    update_book_by_id,
    delete_book_by_id,
    add_author,
    delete_author_by_id, get_author_by_id, get_books_author_by_id,
)
from schemas import BookSchema, AuthorSchema

app = Flask(__name__)
api = Api(app)


class BookList(Resource):
    def get(self) -> tuple[list[dict], int]:

        schema = BookSchema()
        return schema.dump(get_all_books(), many=True), 200

    def post(self) -> tuple[dict, int]:
        data = request.json
        schema = BookSchema()
        try:
            book = schema.load(data)
        except ValidationError as exc:
            return exc.messages, 400

        author = get_author_by_id(book.author_id)
        if author is None:
            return {"message": "Author not found"}, 404

        book = add_book(book)
        return schema.dump(book), 201


class BookItem(Resource):
    def get(self, book_id: int) -> Union[dict, tuple[dict, int]]:
        book = get_book_by_id(book_id)
        if book is None:
            return {'message': 'Book not found'}, 400
        schema = BookSchema()
        return schema.dump(book), 200

    def put(self, book_id) -> Union[dict, tuple[dict, int]]:
        data = request.json
        schema = BookSchema()

        book = get_book_by_id(book_id)
        if book is None:
            return {'message': 'Book not found'}, 400
        try:
            book_data = schema.load(data)
        except ValidationError as exc:
            return exc.messages, 400

        book.title = book_data.title
        book.author_id = book_data.author_id

        update_book_by_id(book)

        return schema.dump(book), 200

    def delete(self, book_id: int) -> tuple[dict, int]:
        book = get_book_by_id(book_id)
        if book is None:
            return {'message': 'Book not found'}, 400

        delete_book_by_id(book_id)

        return {'message': 'Book delete successfully'}, 200


class AuthorList(Resource):
    def post(self) -> tuple[dict, int]:
        data = request.json
        schema = AuthorSchema()
        try:
            author = schema.load(data)
        except ValidationError as exc:
            return exc.messages, 400

        author = add_author(author)
        return schema.dump(author), 201


class AuthorItem(Resource):
    def get(self, author_id: int) -> Union[dict, tuple[dict, int]]:
        author = get_author_by_id(author_id)
        if author is None:
            return {'message': 'Author not found'}, 404

        books = get_books_author_by_id(author_id)
        schema = AuthorSchema()
        author_data = schema.dump(author)
        author_data['books'] = [book.__dict__ for book in books]

        return author_data, 200

    def delete(self, author_id: int) -> tuple[dict, int]:
        author = get_author_by_id(author_id)
        if author is None:
            return {'message': 'Author not found'}, 404
        delete_author_by_id(author_id)
        return {'message': 'Author deleted successfully'}, 204


swagger_authors = Swagger(app, template_file='docs/bookstore.yml')

api.add_resource(BookList, '/api/books')
api.add_resource(BookItem, '/api/books/<int:book_id>')
api.add_resource(AuthorList, '/api/authors')
api.add_resource(AuthorItem, '/api/authors/<int:author_id>')

if __name__ == '__main__':
    init_db(initial_records=DATA)
    WSGIRequestHandler.protocol_version = "HTTP/1.1"
    app.run(debug=True)