REST API w Pythonie: Flask czy FastAPI?
Tworzenie aplikacji internetowych, a w tym REST API, to chleb powszedni backend developerów. Dlatego praca z frameworkiem webowym powinna być szybka i prosta. Microframeworki to bardzo dobry start dla małych projektów, MVP czy nawet dużych aplikacji, które potrzebują REST API – a do nich zaliczają się m.in.: Flask i FastAPI.
Flask jest jedną z najpopularniejszych bibliotek do tworzenia aplikacji internetowych w Pythonie. Osoby zaczynające swoją przygodę z programowaniem bez trudu znajdą na jego temat mnóstwo tutoriali i rozwiązań typowych problemów. Jest on lekki (“microframework”) i bardzo dobrze udokumentowany. Posiada wiele rozszerzeń i sporą społeczność.
FastAPI robi się coraz bardziej popularny z dnia na dzień. Jego nacisk na szybkość (FastAPI), nie tylko w kwestii ilości obsługiwanych zapytań na sekundę, ale również na szybkość developmentu oraz wbudowaną walidację danych – tworzy z niego idealnego kandydata na backendową stronę naszej aplikacji internetowej.
Napisałem aplikację do tworzenia, aktualizowania, pobierania oraz usuwania wiadomości prasowych w dwóch wyżej wymienionych frameworkach i bardzo chętnie przedstawię Wam ich porównanie.
Pierwszą znaczącą różnicą, którą możemy znaleźć przyglądając się tym dwóm bibliotekom to:
Spis treści
Walidacja danych
Instalując Flaska nie dostajemy żadnego narzędzia do walidacji danych. Możemy jednak poradzić sobie przy użyciu dodatków, które oferuje społeczność, np. Flask-Marshmallow czy Flask-Inputs. Minusem tego rozwiązania jest fakt, że musimy polegać na bibliotekach rozwijanych oddzielnie niż nasz główny framework i nie mamy stuprocentowej pewności, że będą one kompatybilne.
FastAPI natomiast daje nam do użytku bibliotekę Pydantic, dzięki której walidacja danych jest o wiele prostsza i szybsza niż pisanie tego z palca. Jest ona ściśle związana z samym FastAPI, więc możemy być pewni, że Pydantic będzie cały czas kompatybilny z naszym frameworkiem.
Jak wyglądają walidacje w poszczególnych bibliotekach na podstawie naszego prostego API?
Tworzymy klasy o nazwach NewsSchema / CreatorSchema, które będą klasami bazowymi do walidacji naszych wiadomości oraz autorów.
|
1 2 3 4 5 6 7 8 9 10 11 |
# Flask @dataclass() class NewsSchema(BaseSchema): title: str = "" content: str = "" creator: CreatorSchema = CreatorSchema() @dataclass class CreatorSchema(BaseSchema): first_name: str = "" last_name: str = "" |
|
1 2 3 4 5 6 7 8 9 |
# FastAPI class NewsSchema(BaseModel): title: str = "" content: str = "" creator: CreatorSchema class CreatorSchema(BaseModel): first_name: str = "" last_name: str = "" |
Możemy zauważyć, że NewsSchema/ CreatorSchema pochodzące z FastAPI używają BaseModel jako klasy nadrzędnej – jest to wymagane, gdyż BaseModel pochodzi z biblioteki Pydantic i posiada niezbędne funkcje do walidacji danych.
Natomiast we Flasku dziedziczymy po klasie BaseSchema, która jest zwykłą dataklasą i zawiera parę metod, z których klasy dziedziczące będą korzystać lub ją nadpisywać. W naszym przypadku sprawdzimy jedynie czy tekst, który wprowadzamy, mieści się w limicie znaków.
Sama walidacja będzie zachodzić w klasach NewsSchemaInput/ CreatorSchemaInput:
|
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 |
# FLASK @dataclass() class NewsSchemaInput(NewsSchema): _errors: dict = field(init=False, default_factory=dict) def _validate_title(self) -> None: if MIN_TITLE_LEN > len(self.title) < MAX_TITLE_LEN: self._errors[ "title" ] = f"Title should be {MIN_TITLE_LEN}-{MAX_TITLE_LEN} characters long" def _validate_content(self) -> None: if len(self.content) < MIN_CONTENT_LEN: self._errors[ "content" ] = f"Content should be minimum {MIN_CONTENT_LEN} characters long" def __post_init__(self) -> None: self._validate_content() self._validate_title() try: if not isinstance(self.creator, CreatorSchemaInput): self.creator = CreatorSchemaInput(**self.creator) except ValidationError as err: self._errors["creator"] = err.errors if self._errors: raise ValidationError( f"Validation failed on {type(self).__name__}", self._errors ) |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# FLASK @dataclass class CreatorSchemaInput(CreatorSchema): _errors: dict = field(init=False, default_factory=dict) def _validate_first_name(self) -> None: if FIRST_NAME_MIN_LEN > len(self.first_name) < FIRST_NAME_MAX_LEN: self._errors[ "first_name" ] = f"First name should be {FIRST_NAME_MIN_LEN}-{FIRST_NAME_MAX_LEN} characters long" def _validate_last_name(self) -> None: if LAST_NAME_MIN_LEN > len(self.last_name) < LAST_NAME_MAX_LEN: self._errors[ "last_name" ] = f"Last name should be {LAST_NAME_MIN_LEN}-{LAST_NAME_MAX_LEN} characters long" def __post_init__(self) -> None: self._validate_first_name() self._validate_last_name() if self._errors: raise ValidationError( f"Validation failed on {type(self).__name__}", self._errors ) |
Gdy stworzymy nasz obiekt NewsSchemaInput/ CreatorSchemaInput, uruchomiona zostanie metoda __post_init__, w której po kolei wywołujemy walidację danych (sprawdzanie długości tekstu) i jeżeli jest ona niepoprawna – dodajemy błędy do zmiennej _errors, a na końcu rzucamy wyjątkiem Validation Error.
W przypadku struktur, które są zagnieżdżone ( CreatorSchemaInput), musimy manualnie tworzyć te obiekty – robimy to po skończonej walidacji NewsSchemaInput w metodzie __post_init__.
Samo sprawdzanie danych nie stanowi większego problemu – dopiero dodawanie nowych pól będzie uciążliwe, ponieważ za każdym razem musimy dodać osobną metodę _validate oraz w przypadku zagnieżdżonej struktury – tworzenie instancji tego obiektu i łapanie wyjątku.
Możemy zauważyć, że klasy, które mają za zadanie walidować przychodzące dane, stają się dosyć rozbudowane – i to tylko dla paru kluczy. Musieliśmy również dodać własną implementację dodawania błędów, dzięki czemu możemy dodawać zagnieżdżone informacje o błędach w odpowiedziach z API.
W FastAPI jest to o wiele prostsze i przyjemniejsze:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
# FastAPI class NewsSchemaInput(NewsSchema): title: str = Field( title="Title of the News", max_length=MAX_TITLE_LEN, min_length=MIN_TITLE_LEN, example="Clickbait title", ) content: str = Field( title="Content of the News", min_length=50, example="Lorem ipsum..." ) creator: CreatorSchemaInput |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# FastAPI class CreatorSchemaInput(CreatorSchema): first_name: str = Field( title="First name of the creator", min_length=FIRST_NAME_MIN_LEN, max_length=FIRST_NAME_MAX_LEN, example="John", ) last_name: str = Field( title="Last name of the creator", min_length=LAST_NAME_MIN_LEN, max_length=LAST_NAME_MAX_LEN, example="Doe", ) |
Poprzez zaimportowanie Field z Pydantic mamy dostęp do prostych deklaracji zasad, które muszą zostać spełnione, aby wprowadzane przez użytkownika dane były prawidłowe.
Również typy danych są walidowane na podstawie typów zmiennych, więc jeżeli nasza zmienna first_name posiada typ str, to musimy w danych wejściowych przekazać tekst (i analogicznie dla wszystkich wbudowanych typów danych).
Bez dodatkowego kodu Pydantic świetnie radzi sobie ze sprawdzaniem zagnieżdżonych struktur (w tym przypadku CreatorSchemaInput). I to wszystko w paru linijkach kodu!
Oprócz max_length i min_length możemy zauważyć też dwa dodatkowe parametry: title oraz example – są one opcjonalne, ale będą widoczne w automatycznej dokumentacji, którą generuje za nas FastAPI.
Serializacja danych wychodzących
Skoro wiemy już, jak walidujemy dane, to powinniśmy zastanowić się, jak chcemy je zwracać. Wiadomość będzie posiadała nie tylko treść, tytuł czy autora, ale również swój unikalny numer (id) oraz datę utworzenia i aktualizacji. Musimy stworzyć nową klasę, która będzie odpowiadała za serializację modelu domenowego News i będzie to NewsSchemaOutput.
|
1 2 3 4 5 6 7 8 9 10 11 12 |
# FLASK @dataclass class NewsSchemaOutput(NewsSchema): id: int = 0 created_at: datetime = datetime.now() updated_at: datetime = datetime.now() def as_dict(self) -> dict: schema_as_dict = super().as_dict() schema_as_dict["created_at"] = int(self.created_at.timestamp()) schema_as_dict["updated_at"] = int(self.updated_at.timestamp()) return schema_as_dict |
|
1 2 3 4 5 6 7 8 |
# FastAPI class NewsSchemaOutput(NewsSchema): id: int = Field(example="26") created_at: datetime = Field(example="1614198897") updated_at: datetime = Field(example="1614198897") class Config: json_encoders = {datetime: lambda dt: int(dt.timestamp())} |
Klasa NewsSchemaOutput w obydwu przypadkach jest praktycznie taka sama, różni się tylko klasą nadrzędną oraz metodą serializacji do słownika (razem ze zmianą obiektu datetime na timestamp).
W FastAPI przy użyciu Pydantic mamy możliwość dodania klasy Config, w której umieściliśmy zmienną json_encoders. Pomaga ona serializować dane w sposób, którego wymagamy. W tym przypadku obiekt daty chcemy przekazać jako timestamp. We Flasku natomiast musieliśmy zmieniać dane w stworzonym już słowniku na takie, które chcemy zwrócić.
Widoki aka endpoints
Tworzenie widoków w obu bibliotekach jest bardzo do siebie podobne i wykorzystuje prosty dekorator na funkcji, której chcemy użyć. Różnią się natomiast sposoby definiowania walidacji danych oraz ich serializacji.
Tworzenie wiadomości, czyli walidacja i serializacja danych
|
1 2 3 4 5 6 7 8 9 |
# FLASK @news_router.route("/news", methods=["POST"]) def add_news(): db_repo = get_database_repo() news_schema = NewsSchemaInput(**request.get_json()) news_dto = NewsDTO.from_news_schema(news_schema=news_schema) saved_news = db_repo.save_news(news_dto=news_dto) output_schema = NewsSchemaOutput.from_entity(news=saved_news).as_dict() return output_schema, HTTPStatus.CREATED |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
# FastAPI @news_router.post( "/news", response_model=NewsSchemaOutput, summary="Create the news", status_code=status.HTTP_201_CREATED, ) async def add_news( news_input: NewsSchemaInput, db_repo: DatabaseRepository = Depends(get_database_repo), ): """ Create the news with following information: - **title**: Title of news - **content**: News content - **creator**: Creator of content """ news_dto = NewsDTO.from_news_schema(news_schema=news_input) db_news = await db_repo.save_news(news_dto=news_dto) return db_news.as_dict() |
Na samym początku mamy dekorator, który ustala ścieżkę dostępu oraz metodę HTTP, która będzie obsługiwana. Flask ustala to za pomocą parametru methods, gdzie musimy przekazać listę obsługiwanych metod, natomiast FastAPI używa atrybutu post na news_router.
Dekorator, którego używa FastAPI, nie służy jedynie do ustalania ścieżki i metod HTTP, ale również do serializacji danych ( response_model), opisania widoku w automatycznej dokumentacji ( summary), zdefiniowania statusu odpowiedzi ( status_code) oraz wielu innych, nie znajdujących się w tym przykładzie. Można powiedzieć, że nie definiuje on jedynie ścieżki dostępu i metody, ale opisuje cały widok w głębi.
No dobrze, ale co się w tym widoku tak naprawdę dzieje?
Zacznijmy od Flaska!
Pierwszą rzeczą, którą robimy, jest pobranie repozytorium bazy danych do naszej funkcji za pomocą:
db_repo = get_database_repo()
W następnym kroku walidujemy przesłane dane przez użytkownika, które znajdują się w obiekcie request:
news_schema = NewsSchemaInput(**request.get_json())
Ta linijka rzuci wyjątkiem ValidationError, jeżeli wprowadzone dane są nieprawidłowe.
Wyjątek zostanie wyłapany (w stworzonym przez nas errorhandler) i Flask zwróci odpowiedź ze wszystkimi błędami, które znajdują się w zmiennej _errors na NewsSchemaInput.
W stworzonym przez nas errorhandler – ale zaraz, zaraz, nie było o tym mowy! We Flasku i FastAPI możemy dodać własne obsługiwanie wyjątków, które zostaną rzucone w implementacji widoków. Wyglądają one tak:
|
1 2 3 4 5 |
# Flask @app.errorhandler(ValidationError) def handle_validation_error(exc: ValidationError) -> Tuple[dict, int]: status_code = HTTPStatus.UNPROCESSABLE_ENTITY return {"detail": exc.errors}, status_code |
|
1 2 3 4 5 6 7 |
# FastAPI @app.exception_handler(ValidationError) async def handle_validation_error(request: Request, exc: ValidationError): return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={"detail": exc.errors()}, ) |
Jeżeli walidacja przebiegła poprawnie, tworzymy obiekt NewsDTO, który przekaże niezbędne informacje do repozytorium bazy danych. Repozytorium wykona swoją magię (zapisze wiadomość do bazy danych) i zwróci nam obiekt domenowy News, który następnie serializujemy za pomocą klasy NewsSchemaOutput:
|
1 2 3 |
news_dto = NewsDTO.from_news_schema(news_schema=news_schema) saved_news = db_repo.save_news(news_dto=news_dto) output_schema = NewsSchemaOutput.from_entity(news=saved_news).as_dict() |
Na samym końcu zwracamy NewsSchemaOutput jako słownik oraz status odpowiedzi.
return output_schema, HTTPStatus.CREATED
Teraz rzućmy okiem na FastAPI.
Tym razem w widoku dostajemy dwa parametry:
news_input oraz
db_repo.
Zacznijmy od tego pierwszego:
Walidacja danych wejściowych dzieje się jeszcze przed uruchomieniem metody naszego widoku, a to dzięki parametrowi news_input.
Zapytacie: ale skąd FastAPI wie, której klasy użyć? Dzięki typowaniu. Parametr news_input posiada typ N