Obsah
1. Rychlá tvorba webových služeb s využitím frameworků FastAPI a SQLAlchemy
2. Příprava projektu s implementací jednoduché webové služby
3. Přidání balíčků s frameworkem FastAPI a Uvicorn
4. Příkaz pro spuštění HTTP serveru s webovou aplikací
5. Realizace koncového bodu / ve vznikajícím REST API webové služby
6. Spuštění webového serveru, poslání požadavku a zobrazení odpovědi
7. Automaticky generované stránky s popisem REST API, zobrazení popisu ve formátu JSON
8. REST API endpointy pro další CRUD operace
9. Webová služba se čtveřicí REST API endpointů
10. Realizace jednoduché databáze se záznamy uloženými v operační paměti
11. Výsledná podoba webové služby s jednoduchou databází realizovanou v paměti
12. Korektní reakce webové služby na chyby – výjimky typu HTTPException
13. Příprava pro připojení webové služby k reálné databázi
15. Připojení k databázi z Pythonu
17. Realizace jednotlivých CRUD operací
18. Výsledná podoba webové služby
19. Repositář s demonstračními příklady
1. Rychlá tvorba webových služeb s využitím frameworků FastAPI a SQLAlchemy
V dnešním článku si ve stručnosti představíme několik balíčků a technologií, které umožňují snadnou a většinou taktéž i velmi rychlou tvorbu webových služeb založených na REST API a taktéž na vybrané (prakticky libovolné) relační databázi. Pro tvorbu těchto služeb použijeme programovací jazyk Python. Konkrétně se zaměříme na praktickou ukázku využití frameworku nazvaného FastAPI (podpora pro webové služby s automatickou tvorbou dokumentace k REST API) zkombinovaného s nástrojem SQLAlchemy (rozhraní k databázové vrstvě). Ovšem interně se budou používat i některé další balíčky určené pro ekosystém programovacího jazyka Python, konkrétně balíček pojmenovaný Uvicorn a taktéž balíčky Pydantic a Starlette.
V navazujících kapitolách si ukážeme postupnou tvorbu jednoduché webové služby, která přes čtyři REST API endpointy nabídne uživatelům této služby čtveřici operací CRUD neboli Create, Read, Update a Delete. Tyto operace budou probíhat nad záznamy uloženými v relační databázi. Konkrétně využijeme databázi PostgreSQL, ale relativně snadno bude možné výsledný projekt upravit takovým způsobem, aby se namísto této databáze použila odlišná relační databáze (od SQLite přes MySQL/MariaDB až po „velké“ databáze typu Oracle). REST API endpointy nabízené námi vytvořenou službou budou popsány s využitím známé specifikace OpenAPI (výsledky jsou dostupné přes Swagger a taktéž přes Redoc), přičemž tato online dokumentace je vygenerována automaticky.
2. Příprava projektu s implementací jednoduché webové služby
V prvním kroku si vytvoříme kostru projektu, který postupně upravíme do formy skutečné webové služby. Pro správu projektů využijeme nástroj PDM, který byl popsán v článku PDM: moderní správce balíčků a virtuálních prostředí Pythonu (ovšem stejně dobře můžete použít i klasický pip zkombinovaný s venv nebo virtualenv). Kostra nového projektu se vytvoří příkazem:
$ pdm init
Po zadání tohoto příkazu se systém PDM zeptá na několik otázek. Odpovědi jsou zvýrazněny tučným písmem:
Creating a pyproject.toml for PDM...
Please enter the Python interpreter to use
0. /usr/bin/python (3.11)
1. /usr/bin/python3.11 (3.11)
2. /usr/bin/python3 (3.11)
Please select (0): 0
Would you like to create a virtualenv with /usr/bin/python? [y/n] (y): y
Virtualenv is created successfully at /home/ptisnovs/test2/.venv
Is the project a library that is installable?
If yes, we will need to ask a few more questions to include the project name and build backend [y/n] (n): n
License(SPDX name) (MIT):
Author name (): Pavel Tisnovsky
Author email (): tisnik@nowhere.us
Python requires('*' to allow any) (>=3.11): >=3.8
Project is initialized successfully
Výsledkem by měla být tato kostra projektu:
.
├── pyproject.toml
├── README.md
├── src
│ └── example_package
│ └── __init__.py
└── tests
└── __init__.py
Projektový soubor pyproject.toml bude mít následující obsah:
[project]
name = ""
version = ""
description = ""
authors = [
{name = "", email = ""},
]
requires-python = ">=3.8"
readme = "README.md"
license = {text = "MIT"}
3. Přidání balíčků s frameworkem FastAPI a Uvicorn
Ve druhém kroku do postupně vznikajícího projektu přidáme balíček nazvaný fastapi společně s balíčkem uvicorn. Pro tento účel se použije následující příkaz:
$ pdm add fastapi uvicorn
Alternativně lze použít dvojici samostatných příkazů, pokud preferujete instalovat balíčky postupně:
$ pdm add uvicorn $ pdm add fastapi
Projektový soubor po provedení těchto kroků doznal změny – viz zvýrazněnou část kódu:
[project]
name = ""
version = ""
description = ""
authors = [
{name = "", email = ""},
]
dependencies = [
"fastapi>=0.104.0",
"uvicorn>=0.23.2",
]
requires-python = ">=3.8"
readme = "README.md"
license = {text = "MIT"}
4. Příkaz pro spuštění HTTP serveru s webovou aplikací
Ve třetím kroku do projektového souboru pyproject.toml přidáme uživatelský příkaz nazvaný start, kterým budeme spouštět HTTP server s webovou službou. Nebude tedy nutné neustále opakovat dlouhý a poměrně těžko zapamatovatelný příkaz; namísto toho bude stačit na příkazové řádce zapsat „pdm start“:
[project]
name = ""
version = ""
description = ""
authors = [
{name = "", email = ""},
]
dependencies = [
"fastapi>=0.104.0",
"uvicorn>=0.23.2",
]
requires-python = ">=3.11"
readme = "README.md"
license = {text = "MIT"}
[tool.pdm.scripts]
start = "uvicorn --app-dir src/example_package main:app --reload"
5. Realizace koncového bodu / ve vznikajícím REST API webové služby
Nyní nadešel čas pro důležitý krok. Musíme totiž vytvořit funkci zavolanou pro obsluhu události, která vznikne ve chvíli, kdy na webovou službu přijde požadavek na koncový bod endpoint /, tedy například požadavek směrovaný na URL http://127.0.0.1:8000/ (pro lokálně běžící HTTP server). Do souboru „src/example_package/main.py“ (jméno je nutné dodržet) zapíšeme programový kód, v němž se inicializuje webová služba nazvaná app a poté se registruje handler pro koncový bod / volaný s HTTP metodou GET (což mj. znamená snadné otestování v prohlížeči). Handler odpoví tak, že pošle klientovi JSON s jediným atributem a hodnotou:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def origin():
return {"message": "My first FastAPI app"}
6. Spuštění webového serveru, poslání požadavku a zobrazení odpovědi
HTTP server s webovou službou spustíme příkazem pdm start:
$ pdm start
Na terminál by se nyní měly vypsat informace o adrese a portu, na kterém HTTP server běží:
INFO: Will watch for changes in these directories: ['/home/ptisnovs/x/app2'] INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: Started reloader process [11351] using StatReload INFO: Started server process [11353] INFO: Waiting for application startup. INFO: Application startup complete.
Takto vypadá odpověď na HTTP požadavek poslaný webovým prohlížečem:
Obrázek 1: Odpověď na HTTP požadavek poslaný webovým prohlížečem.
Pochopitelně můžeme s webovou službou komunikovat i přes curl či podobné nástroje (požadavek je uvozen znakem <, hlavička odpovědi znakem >):
$ curl -v localhost:8000
* Trying 127.0.0.1:8000...
* Connected to localhost (127.0.0.1) port 8000 (#0)
> GET / HTTP/1.1
> Host: localhost:8000
> User-Agent: curl/8.0.1
> Accept: */*
>
< HTTP/1.1 200 OK
< date: Tue, 07 Nov 2023 08:43:30 GMT
< server: uvicorn
< content-length: 34
< content-type: application/json
<
* Connection #0 to host localhost left intact
{"message":"My first FastAPI app"}
7. Automaticky generované stránky s popisem REST API, zobrazení popisu ve formátu JSON
Po spuštění webového serveru nebude ve skutečnosti dostupný pouze koncový bod /, ale i další adresy se speciálním významem. Jedná se zejména o adresu localhost:8000/doc, na níž je umístěna stránka s interaktivním formulářem, který umožňuje si všechny koncové body REST API otestovat. Pokud znáte Swagger, nebude pro vás obsah této stránky překvapující:
Obrázek 2: Interaktivní formulář, který umožňuje si otestovat všechny koncové body REST API.
Obrázek 3: Otestování koncového bodu / na interaktivním formuláři.
Na adrese localhost:8080/redoc nalezneme podobný formulář, pouze s odlišným designem a nepatrně odlišným chováním:
Obrázek 4: Alternativní interaktivní formulář, který umožňuje si otestovat všechny koncové body realizované ve webové službě.
Obrázek 5: Otestování koncového bodu / na (alternativním) interaktivním formuláři.
A kromě toho si můžeme zobrazit popis REST API realizovaný podle standardu OpenAPI. Po naformátování vypadá popis REST API s jediným koncovým bodem následovně:
{
"openapi": "3.1.0",
"info": {
"title": "FastAPI",
"version": "0.1.0"
},
"paths": {
"/": {
"get": {
"summary": "Origin",
"operationId": "origin__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
}
}
}
}
}
}
8. REST API endpointy pro další CRUD operace
V dalším kroku do naší postupně vznikající webové služby přidáme další REST API endpointy pro zbývající tři CRUD operace. Připomeňme si, že CRUD je zkratka znamenající CREATE, READ, UPDATE a DELETE. Naše koncové body by měly být navrženy následovně:
| Operace | Endpoint | HTTP metoda | Stručný popis |
|---|---|---|---|
| CREATE | /create | POST | vytvoří nový záznam, parametrem bude text, který se v záznamu uloží |
| READ | / | GET | vrátí všechny záznamy |
| UPDATE | /update/{id} | PUT | změní text záznamu identifikovaný jeho ID |
| DELETE | /delete/{id} | DELETE | vymaže záznam identifikovaný jeho ID |
9. Webová služba se čtveřicí REST API endpointů
Druhá verze naší webové služby bude obsahovat definice handlerů pro všechny čtyři REST API endpointy. Povšimněte si, jak se s využitím dekorátorů specifikují jak HTTP metody, tak i adresy endpointů. Parametry posílané v požadavku (resp. očekávané v požadavku) jsou odvozeny od názvů a typů parametrů jednotlivých handlerů. To vlastně znamená, že žádné další informace nemusíme frameworku FastAPI předávat – vše je specifikováno na úrovni zdrojového kódu Pythonu s využitím typových informací (type hints):
from fastapi import FastAPI
app = FastAPI()
@app.post("/create")
async def create_operation(text: str):
return {"created": text}
@app.get("/")
async def read_operation():
return {"list": None}
@app.put("/update/{id}")
async def update_operation(id: int, text: str = ""):
pass
@app.delete("/delete/{id}")
async def delete_operation(id: int):
return {"deleted": id}
Po spuštění této varianty webové služby si můžeme nechat zobrazit interaktivní formulář s popisem všech endpointů REST API:
Obrázek 6: Nově definované endpointy jsou automaticky zobrazeny na tomto formuláři, kde si je lze otestovat.
10. Realizace jednoduché databáze se záznamy uloženými v operační paměti
Ve finální verzi naší webové aplikace použijeme reálnou databázi, ovšem prozatím si vystačíme s velmi jednoduchou databází uloženou pouze v operační paměti. Tato databáze bude realizována instancí třídy, v níž je jako atribut uložen slovník se záznamy a taktéž atribut obsahující ID naposledy uloženého záznamu (budeme tak simulovat tabulku s dvojicí sloupců s ID a textem, přičemž ID se při zápisu dalšího záznamu zvyšuje). Samozřejmě se nejedná ani o nejlepší ani o nejvýkonnější řešení, ale pro ilustraci realizovaných operací může být zpočátku dostačující:
class Storage():
def __init__(self):
self._items = {}
self._id = 0
def create(self, text):
self._id += 1
self._items[self._id] = text
return self._id
def read(self):
return self._items
def update(self, id, text):
self._items[id] = text
def delete(self, id):
del self._items[id]
11. Výsledná podoba webové služby s jednoduchou databází realizovanou v paměti
Nyní již můžeme upravit handlery pro jednotlivé koncové body tak, že se do výše popsané databáze budou ukládat nové záznamy, bude se číst seznam záznamů, záznamy bude možné upravit a taktéž je mazat – budu tedy realizovány všechny čtyři CRUD operace. Pokud nebudeme chtít provádět další kontroly (například na existenci záznamu), může naše webová služba vypadat takto:
from fastapi import FastAPI
storage = Storage()
app = FastAPI()
@app.post("/create")
async def create_operation(text: str):
id = storage.create(text)
return {"created": id}
@app.get("/")
async def read_operation():
return {"list": storage.read()}
@app.put("/update/{id}")
async def update_operation(id: int, text: str = ""):
storage.update(id, text)
return {"updated": id, "new text": text}
@app.delete("/delete/{id}")
async def delete_operation(id: int):
storage.delete(id)
return {"deleted": id}
Obrázek 7: Endpointy nyní již pracují podle očekávání.
Pro zajímavost se podívejme, jak vypadá automaticky vygenerovaný popis celého REST API:
{
"openapi": "3.1.0",
"info": {
"title": "FastAPI",
"version": "0.1.0"
},
"paths": {
"/create": {
"post": {
"summary": "Create Operation",
"operationId": "create_operation_create_post",
"parameters": [
{
"name": "text",
"in": "query",
"required": true,
"schema": {
"type": "string",
"title": "Text"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/": {
"get": {
"summary": "Read Operation",
"operationId": "read_operation__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
}
}
}
},
"/update/{id}": {
"put": {
"summary": "Update Operation",
"operationId": "update_operation_update__id__put",
"parameters": [
{
"name": "id",
"in": "path",
"required": true,
"schema": {
"type": "integer",
"title": "Id"
}
},
{
"name": "text",
"in": "query",
"required": false,
"schema": {
"type": "string",
"default": "",
"title": "Text"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/delete/{id}": {
"delete": {
"summary": "Delete Operation",
"operationId": "delete_operation_delete__id__delete",
"parameters": [
{
"name": "id",
"in": "path",
"required": true,
"schema": {
"type": "integer",
"title": "Id"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail"
}
},
"type": "object",
"title": "HTTPValidationError"
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "integer"
}
]
},
"type": "array",
"title": "Location"
},
"msg": {
"type": "string",
"title": "Message"
},
"type": {
"type": "string",
"title": "Error Type"
}
},
"type": "object",
"required": [
"loc",
"msg",
"type"
],
"title": "ValidationError"
}
}
}
}
12. Korektní reakce webové služby na chyby – výjimky typu HTTPException
Operace UPDATE a DELETE očekávají, že se jim předá ID upravovaného nebo mazaného prvku. Co se však stane resp. má stát v případě, že prvek s tímto ID neexistuje? Webová služba by měla reagovat tak, že vrátí odpověď s vhodným stavovým kódem protokolu HTTP doplněným o informaci, k jaké chybě došlo. Reakce na neexistující prvek by mohla vést k odpovědi se stavovým kódem 404 Not Found (po 200 OK asi nejznámější stavový kód). V případě frameworku FastAPI nám postačuje z handleru vyhodit výjimku typu HTTPException, při jejíž konstrukci se specifikuje jak stavový kód, tak i chybové hlášení.
Naši webovou službu nepatrně upravíme následujícím způsobem:
from fastapi import FastAPI
from fastapi import HTTPException
class Storage():
def __init__(self):
self._items = {}
self._id = 0
def create(self, text):
self._id += 1
self._items[self._id] = text
return self._id
def read(self):
return self._items
def update(self, id, text):
# test???
self._items[id] = text
def delete(self, id):
del self._items[id]
storage = Storage()
app = FastAPI()
@app.post("/create")
async def create_operation(text: str):
id = storage.create(text)
return {"created": id}
@app.get("/")
async def read_operation():
return {"list": storage.read()}
@app.put("/update/{id}")
async def update_operation(id: int, text: str = ""):
try:
storage.update(id, text)
return {"updated": id, "new text": text}
except:
raise HTTPException(status_code=404, detail="Item not found")
@app.delete("/delete/{id}")
async def delete_operation(id: int):
try:
storage.delete(id)
return {"deleted": id}
except:
raise HTTPException(status_code=404, detail="Item not found")
Výsledek si pochopitelně opět můžeme velmi snadno otestovat:
Obrázek 8: Nyní mohou různé požadavky vracet stavové kódy protokolu HTTP s podrobnějšími informacemi o chybě.
13. Příprava pro připojení webové služby k reálné databázi
Finální verze naší webové služby bude využívat reálnou databázi, konkrétně PostgreSQL. Pro manipulaci se záznamy uloženými v databázi použijeme objektově relační mapování (ORM), konkrétně projekt SQLAlchemy. Do našeho projektu tedy bude nutné přidat závislost právě na SQLAlchemy. Ovšem kromě toho musíme nainstalovat i balíček s rozhraním k databázi (někdy se označuje slovem ovladač, driver). Tento balíček se jmenuje psycopg2-binary. Oba zmíněné balíčky nainstalujeme následujícím způsobem:
$ pdm add sqlalchemy $ pdm add psycopg2-binary
Projektový soubor pyproject.toml by měl vypadat zhruba následovně (verze některých balíčků se ovšem mohou o setiny lišit):
[project]
name = ""
version = ""
description = ""
authors = [
{name = "", email = ""},
]
dependencies = [
"fastapi>=0.104.0",
"uvicorn>=0.23.2",
"sqlalchemy>=2.0.22",
"psycopg2-binary>=2.9.9",
]
requires-python = ">=3.11"
readme = "README.md"
license = {text = "MIT"}
[tool.pdm.scripts]
start = "uvicorn --app-dir src/example_package main:app --reload"
14. Konfigurace databáze
Nastává potenciálně nejsložitější krok – instalace, konfigurace a zprovoznění databáze PostgreSQL. Samotná instalace databáze je jednoduchá, protože PostgreSQL je dostupná jako standardní balíček v prakticky každé distribuci Linuxu. Například u distribucí založených na Debianu se instalace provede následovně:
$ sudo apt install postgresql postgresql-contrib
Poté musíme zajistit, aby byla databáze spuštěna:
$ sudo systemctl start postgresql.service
nebo alternativně:
$ sudo service postgresql start
To ovšem není vše, protože musíme v Postgresu vytvořit nového uživatele (říkejme mu tester), nastavit mu heslo a nakonec vytvořit databázi, ke které bude mít veškerá potřebná práva. Přihlásíme se do konzole jako systémový uživatel postgres (ten byl vytvořen při instalaci Postgresu):
$ sudo -i -u postgres
Vytvoříme nového uživatele a specifikujeme jeho heslo:
postgres=# CREATE USER tester WITH PASSWORD '123qwe'; CREATE ROLE
Dále vytvoříme databázi a zajistíme, aby k ní měl právě vytvořený uživatel práva:
postgres=# CREATE DATABASE test1 OWNER tester; CREATE DATABASE postgres=# GRANT ALL PRIVILEGES ON DATABASE test1 TO tester; GRANT
Nyní by mělo být možné se přihlásit jako uživatel tester do Postgresu přes její standardní konzoli:
$ psql -U tester
Pokud se hlásí problémy při přihlašování, je vhodné se ujistit, jak jsou nakonfigurovány parametry pro připojení k databázi. Viz například článek na adrese https://tecadmin.net/postgresql-allow-remote-connections/ (zejména jeho druhá kapitola).
V konzoli Postgresu si necháme zobrazit informaci o dostupných databázích:
tester=> \l
List of databases
Name | Owner | Encoding | Collate | Ctype | ICU Locale | Locale Provider | Access privileges
-----------+----------+----------+-------------+-------------+------------+-----------------+-----------------------
postgres | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc |
template0 | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc | =c/postgres +
| | | | | | | postgres=CTc/postgres
template1 | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc | =c/postgres +
| | | | | | | postgres=CTc/postgres
test1 | tester | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc | =Tc/tester +
| | | | | | | tester=CTc/tester
(4 rows)
Databáze test1 by měla být prozatím prázdná:
tester=> \c test1 You are now connected to database "test1" as user "tester". tester=> \dt+ Did not find any relations.
15. Připojení k databázi z Pythonu
V případě, že databáze běží, je dostupná (pro připojení se k ní) a existuje v ní uživatel tester se zadaným heslem, je realizace připojení k takové databázi relativně triviální. V případě, že nebudeme načítat parametry připojení z proměnných prostředí (k čemuž dříve či později pravděpodobně dojdete), zajistí připojení funkce connect_to_db, která vrací dvojici objektů (instance DB enginu a objekt s informacemi o realizovaném sezení). Pochopitelně může nastat mnoho situací, kdy připojení není možné navázat, uživatel nemá právo k připojení k databázi, jeho heslo je neplatné, databáze test1 neexistuje atd. V takovém případě tato funkce vyhodí výjimku v době inicializace webové služby:
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.orm import sessionmaker
def connect_to_db():
url = URL.create(
drivername="postgresql",
username="tester",
password="123qwe",
host="localhost",
database="test1",
port=5432
)
print("url", url)
engine = create_engine(url)
print("engine", engine)
Session = sessionmaker(bind=engine)
session = Session()
print("session", session)
return engine, session
engine, session = connect_to_db()
16. Deklarace modelu pro ORM
Jak jsme si již řekli v předchozích kapitolách, budeme pro práci se záznamy uloženými do databáze používat objektově-relační mapování neboli ORM. Pro programovací jazyk Python existuje několik balíčků s implementací ORM; my dnes použijeme SQLAlchemy, což je (pravděpodobně) nejznámější balíček z této kategorie.
Záznamy, které budou zapisovány do databáze či naopak čteny z databáze, jsou popsány takzvaným modelem. Jedná se vlastně o spojení objektového modelu (třídy) s metadaty, které určují, jak se budou instance této třídy ukládat do databáze. Vzhledem k tomu, že nebudeme používat žádné složitější databázové schéma (mapování 1:N, M:N atd.), může být model mnohdy i poměrně triviální.
V naší konkrétní webové službě budeme pracovat se záznamy, které obsahují dvojici atributů nazvaných id a text. Záznamy budou ukládány do tabulky nazvané records, která bude obsahovat dva sloupce pojmenované id a text, přičemž id je automaticky generovaným primárním klíčem. Typ hodnot i další metadata sloupců jsou popsány instancí třídy Column. Celý model a vlastně i ORM mapování je tak popsáno na několika řádcích Pythonovského kódu:
Base = declarative_base()
class Record(Base):
__tablename__ = "records"
id = Column(Integer, primary_key=True)
text = Column(String)
# inicializace
Base.metadata.create_all(engine)
Poslední řádek zajistí vytvoření tabulky records v případě, že tato tabulka ještě neexistuje.
17. Realizace jednotlivých CRUD operací
Podívejme se nyní na jeden možný způsob realizace handlerů pro jednotlivé CRUD operace. Nejjednodušší je v tomto případě operace CREATE, která bude interně provedena SQL příkazem INSERT. Vytvoříme novou instanci třídy Record a metodou Session.add ji zapíšeme do databáze. Tím se automaticky upraví i atribut id, který vrátíme klientovi:
record = Record(text=text)
session.add(record)
session.commit()
return {"created": record.id}
Operace READ je realizována SQL příkazem SELECT * from records. V Pythonu ji zapíšeme takto (samozřejmě v případě zápisu podmínky WHERE, operace řazení, operace seskupení atd. bude situace nepatrně složitější):
record_query = session.query(Record) records = record_query.all()
Operace UPDATE a DELETE jsou realizovány SQL příkazy se stejným jménem. Zde se vyžaduje specifikace konkrétního záznamu, který se má upravit či smazat. Záznamy budeme vybírat přes jejich ID, takže interně se do SQL příkazů UPDATE a DELETE vloží příslušná klauzule WHERE:
# UPDATE record_query = session.query(Record).filter(Record.id==id) record = record_query.first() record.text = text session.add(record) session.commit() # DELETE record = session.query(Record).filter(Record.id==id).first() session.delete(record) session.commit()
Konkrétně budou handlery všech čtyř CRUD operací vypadat následovně:
app = FastAPI()
@app.post("/create")
async def create_operation(text: str):
record = Record(text=text)
session.add(record)
session.commit()
return {"created": record.id}
@app.get("/")
async def read_operation():
record_query = session.query(Record)
return {"list": record_query.all()}
@app.put("/update/{id}")
async def update_operation(id: int, text: str = ""):
record_query = session.query(Record).filter(Record.id==id)
record = record_query.first()
record.text = text
session.add(record)
session.commit()
@app.delete("/delete/{id}")
async def delete_operation(id: int):
record = session.query(Record).filter(Record.id==id).first()
session.delete(record)
session.commit()
return {"todo deleted": record.id}
Při prvním spuštění takto upravené webové služby se tabulka records vytvoří, což si můžeme snadno ověřit v konzoli psql:
tester=> \l
List of databases
Name | Owner | Encoding | Collate | Ctype | ICU Locale | Locale Provider | Access privileges
-----------+----------+----------+-------------+-------------+------------+-----------------+-----------------------
postgres | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc |
template0 | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc | =c/postgres +
| | | | | | | postgres=CTc/postgres
template1 | postgres | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc | =c/postgres +
| | | | | | | postgres=CTc/postgres
test1 | tester | UTF8 | en_US.UTF-8 | en_US.UTF-8 | | libc | =Tc/tester +
| | | | | | | tester=CTc/tester
(4 rows)
tester=> \c test1
You are now connected to database "test1" as user "tester".
test1=> \dt+
List of relations
Schema | Name | Type | Owner | Persistence | Access method | Size | Description
--------+---------+-------+--------+-------------+---------------+-------+-------------
public | records | table | tester | permanent | heap | 16 kB |
(1 row)
test1=> select * from records;
id | text
----+--------------
1 | prvni zaznam
2 | dalsi zaznam
(2 rows)
18. Výsledná podoba webové služby
Všechny části zdrojových kódů popsaných v předchozích třech kapitolách lze nyní spojit a vytvořit tak výslednou podobu webové služby. Její zdrojový kód bude vypadat následovně:
from fastapi import FastAPI
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.orm import sessionmaker
def connect_to_db():
url = URL.create(
drivername="postgresql",
username="tester",
password="123qwe",
host="localhost",
database="test1",
port=5432
)
print("url", url)
engine = create_engine(url)
print("engine", engine)
Session = sessionmaker(bind=engine)
session = Session()
print("session", session)
return engine, session
engine, session = connect_to_db()
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class Record(Base):
__tablename__ = "records"
id = Column(Integer, primary_key=True)
text = Column(String)
# inicializace
Base.metadata.create_all(engine)
app = FastAPI()
@app.post("/create")
async def create_operation(text: str):
record = Record(text=text)
session.add(record)
session.commit()
return {"created": record.id}
@app.get("/")
async def read_operation():
record_query = session.query(Record)
return {"list": record_query.all()}
@app.put("/update/{id}")
async def update_operation(id: int, text: str = ""):
record_query = session.query(Record).filter(Record.id==id)
record = record_query.first()
record.text = text
session.add(record)
session.commit()
@app.delete("/delete/{id}")
async def delete_operation(id: int):
record = session.query(Record).filter(Record.id==id).first()
session.delete(record)
session.commit()
return {"todo deleted": record.id}
Pro rozsáhlejší webové služby může být výhodnější zdrojový kód vhodným způsobem rozdělit, například tak, aby to odpovídalo nějakému vzoru (MVC atd.). Ovšem pro službu se čtyřmi triviálními CRUD operacemi to (podle mého skromného názoru) ani není nutné.
Webovou službu si pochopitelně můžeme otestovat – přes endpoint /create do databáze zapíšeme záznamy, které potom přečteme přes endpoint / popř. upravíme nebo vymažeme přes endpointy /update a /delete:
Obrázek 9: Nyní by všechny čtyři endpointy měly být funkční.
19. Repositář s demonstračními příklady
Všech šest projektů webových služeb, které jsme si ukázali v dnešním článku, naleznete na adrese https://github.com/tisnik/most-popular-python-libs. Následují odkazy na jednotlivé příklady (pro jejich spuštění je pochopitelně nutné mít nainstalovánu knihovnu FastAPI a její závislosti):
| # | Příklad | Adresa | |
|---|---|---|---|
| 1 | app1 | kostra projektu, na kterém se bude stavět webová služba | https://github.com/tisnik/most-popular-python-libs/blob/master/fastapi/app1 |
| 2 | app2 | webová služba s jediným REST API endpointem / | https://github.com/tisnik/most-popular-python-libs/blob/master/fastapi/app2 |
| 3 | app3 | webová služba se čtveřicí REST API endpointů | https://github.com/tisnik/most-popular-python-libs/blob/master/fastapi/app3 |
| 4 | app4 | implementace jednoduché in-memory databáze | https://github.com/tisnik/most-popular-python-libs/blob/master/fastapi/app4 |
| 5 | app5 | využití výjimek typu HTTPException | https://github.com/tisnik/most-popular-python-libs/blob/master/fastapi/app5 |
| 6 | app6 | realizace CRUD operací pro PostgreSQL | https://github.com/tisnik/most-popular-python-libs/blob/master/fastapi/app6 |
20. Odkazy na Internetu
- FastAPI
https://fastapi.tiangolo.com/ - Pydantic
https://docs.pydantic.dev/latest/ - What does async actually do in FastAPI?
https://stackoverflow.com/questions/75463993/what-does-async-actually-do-in-fastapi - FastAPI runs api-calls in serial instead of parallel fashion
https://stackoverflow.com/questions/71516140/fastapi-runs-api-calls-in-serial-instead-of-parallel-fashion - SQLAlchemy
https://www.sqlalchemy.org/ - PostgreSQL
https://www.postgresql.org/ - OpenAPI Specification
https://github.com/OAI/OpenAPI-Specification - Swagger
https://swagger.io/ - Redoc
https://github.com/Redocly/redoc - PDM: moderní správce balíčků a virtuálních prostředí Pythonu
https://www.root.cz/clanky/pdm-moderni-spravce-balicku-a-virtualnich-prostredi-pythonu/ - Stránka projektu PDM
https://pdm.fming.dev/latest/ - PDM na GitHubu
https://github.com/pdm-project/pdm - PEP 582 – Python local packages directory
https://peps.python.org/pep-0582/ - PDM na PyPi
https://pypi.org/project/pdm/ - Which Python package manager should you use?
https://towardsdatascience.com/which-python-package-manager-should-you-use-d0fd0789a250 - How to Use PDM to Manage Python Dependencies without a Virtual Environment
https://www.youtube.com/watch?v=qOIWNSTYfcc - What are the best Python package managers?
https://www.slant.co/topics/2666/~best-python-package-managers - HTTP PUT vs. POST
https://restfulapi.net/rest-put-vs-post/
ČLÁNKY DO MAILU