FastAPI ve Raw SQL ile RestAPI Projesi¶
Bu yazıda, psycopg2 kütüphanesi ile raw(ham) SQL sorguları kullanarak bir FastAPI CRUD (Create/Oluştur, Read/Oku, Update/Güncelle, Delete/Sil) uygulamasının nasıl oluşturulacağını öğreneceğiz.
Bir FastAPI uygulamasında ham SQL kullanmanın ORM kütüphanelerini kullanmanın avantajlarını ve dezavantajlarını karşılaştıracağız. Bu makalenin sonunda, FastAPI'de ham SQL'den nasıl yararlanacağınız konusunda temel bir bilgiye sahip olacak ve bunun projeniz için doğru yaklaşım olup olmadığı konusunda bilinçli bir karar vereceksiniz.
İçindekiler:
FastAPI ve psycopg2'ye Giriş Ham SQL'in Artıları ve Eksileri Başlarken: Projeyi Kurma PostgreSQL Veritabanına Bağlanmak Ham SQL ile CRUD Uç Noktalarını Oluşturma Hata İşleme(Error Handling) ve Veri Doğrulama(Data Validation) FastAPI CRUD Uygulamasını Test Etme Sonuç
Part 1¶
1. FastAPI ve psycopg2'ye Giriş¶
FastAPI, mükemmel performansı ve kullanım kolaylığı nedeniyle popülerlik kazanmış, Python ile API'ler oluşturmaya yönelik modern, hızlı bir web frameworktür. Son derece verimli kod ve otomatik döküman oluşturma sağlamak için Python 3.7+ veri tipi tanımlama ve diğer modern özelliklerden yararlanır. FastAPI, güçlü bir ASGI framework olan Starlette üzerine kuruludur ve bu da onun yüksek veri yüklerini işlemesine ve etkili bir şekilde ölçeklenmesine olanak tanır.
FastAPI'de veritabanlarıyla etkileşim söz konusu olduğunda, yaygın bir yaklaşım PostgreSQL veritabanı yönetimi için psycopg2 kütüphanesini kullanmaktır. Psycopg2, PostgreSQL veritabanlarıyla çalışmanın basit ve verimli bir yolunu sağlayan, PostgreSQL için yaygın olarak kullanılan ve köklü bir Python bağdaştırıcısıdır(adapter).
2. Ham SQL'in Artıları ve Eksileri:¶
Ham SQL, SQL sorgularını Nesne-İlişkisel Eşleme (Object Relational Mapping, ORM) kütüphaneleri kullanmadan doğrudan uygulama kodunuza yazma pratiği anlamına gelir. Bir FastAPI uygulamasında ham SQL kullanmanın bazı artılarını ve eksilerini inceleyelim:
Artıları¶
Esneklik: Ham SQL, sorgularınız üzerinde tam kontrol sağlayarak veritabanına özgü özelliklerden ve optimizasyonlardan yararlanmanıza olanak tanır. Özel ihtiyaçlarınıza göre uyarlanmış karmaşık sorgular yazabilirsiniz. Performans: Optimize edilmiş SQL sorguları oluşturarak, ORM kütüphanelerini kullanmaya kıyasla genellikle daha iyi performans elde edebilirsiniz. Çalıştırma üzerinde doğrudan kontrole sahipsiniz ve daha iyi verimlilik için ince ayar yapabilirsiniz. Geçiş: Veritabanı şemanız değiştiğinde, yürütülen SQL ifadeleri üzerinde doğrudan denetime sahip olduğunuz için ham SQL kullanmak geçiş komut dosyalarının uygulanmasını kolaylaştırabilir.
Eksileri¶
Kod Bakımı: Uygulama kodunuza ham SQL sorguları yazmak, özellikle uygulamanızın karmaşıklığı arttıkça iş mantığını sürdürmeyi ve anlamayı zorlaştırabilir. Standart Kod: Ham SQL, sorgu sonuçları ile Python nesneleri arasındaki eşlemeyi manuel olarak işlemek için daha fazla kod gerektirir. Bu, ek geliştirme çabalarına ve potansiyel hata kaynaklarına neden olabilir. Taşınabilirlik: Ham SQL kullanmak, uygulamanızı belirli bir veritabanı sistemine yakından bağlar ve bu durumda farklı bir veritabanına geçmeyi zorlaştırır.
3. Başlarken: Projeyi Kurma¶
news_fastapi klasörü oluşturun ve favori kod editörünüzü açın. Terminalinizi açın ve news_fastapi dizinine gidin. Python projeleri için sanal ortam oluşturulması tavsiye edilir.
$ cd news_fastapi
# venv modülünü(-m) kullanarak sanal ortam oluşturun ve venv olarak adlandırın
$ python3.11 -m venv venv
# venv aktif edelim
$ source venv/bin/activate
# windows kullanıcıları aşağıdaki gibi etkinleştirebilir
$ venv\Scripts\activate
from fastapi import FastAPI
app = FastAPI()
@app.get('/')
async def index():
return {'message': 'Welcome to news app!'}
- fastapi'den import FastAPI: Bu satır, FastAPI sınıfını fastapi modülünden içe aktarır.
- FastAPI, uygulamamızı oluşturmak için kullanacağımız ana sınıftır.
- app = FastAPI(): Bu satır, FastAPI sınıfının bir örneğini oluşturur ve onu app değişkenine atar. Rotalarımızı tanımlamak ve uygulamayı çalıştırmak için bu uygulama nesnesini kullanacağız.
- @app.get('/'): Bu, FastAPI'ye root path'e ("/") bir GET isteği yapıldığında aşağıdaki fonksiyonun çalıştırılması gerektiğini söyleyen bir dekoratördür.
- async def index(): Bu, index fonksiyonunu asenkron bir fonksiyon olarak tanımlar. async anahtar sözcüğü, fonksiyon içinde await kullanmamızı sağlar.
- return {'message': 'Welcome to news app!'}: Bu satır, anahtarın "message" ve değerinin "Welcome to news app!" olduğu tek bir anahtar-değer çifti içeren bir sözlük(dict) döndürür. FastAPI, sözlüğü otomatik olarak JSON biçiminde serialize eder ve yanıt olarak gönderir. Özetlemek gerekirse, bu kod, bir GET isteği aracılığıyla erişildiğinde bir JSON mesajıyla yanıt veren tek bir rotaya (/) sahip bir FastAPI uygulaması oluşturur. Mesaj basit bir selamlamadır, "Welcome to news app"
Uygulamayı çalıştıralım
- uvicorn app.main:app komutu şunu ifade eder:-
main: app klasörü altındaki main.py dosyası (Python "modülü").
-
app: main.py içinde app = FastAPI() satırıyla oluşturulan nesne.
-
--reload: kod değişikliklerinden sonra sunucunun yeniden başlatılmasını sağlar. Sadece geliştirme/development için kullanın.
-
--workers 1 : workers miktarı
-
--host : host ip adresi
-
--port : port numarası Tarayıcınızı açın ve http://0.0.0.0:8000/ adresine gidin. {'message': 'Welcome to news app!'} json yanıt mesajını göreceksiniz.
4. PostgreSQL Veritabanına Bağlanma¶
.env dosyasından ortam değişkenlerini okumak için zaten uvicorn ile kurulu olan python-dotenv paketini kullanacağım
app klasörü altında database.py dosyası oluşturun.
import os
from pathlib import Path
import dotenv
BASE_DIR = Path(__file__).resolve().parent.parent
dotenv.load_dotenv(BASE_DIR / ".env")
import os
from pathlib import Path
import dotenv
from abc import ABC, abstractmethod # new
BASE_DIR = Path(__file__).resolve().parent.parent
dotenv.load_dotenv(BASE_DIR / ".env")
class Database(ABC):
"""
Database context manager
"""
def __init__(self, driver) -> None:
self.driver = driver
@abstractmethod
def connect_to_database(self):
raise NotImplementedError()
def __enter__(self):
self.connection = self.connect_to_database()
self.cursor = self.connection.cursor()
return self
def __exit__(self, exception_type, exc_val, traceback):
self.cursor.close()
self.connection.close()
-
__init__
fonksiyonu, Database sınıfının yapıcısıdır. Veritabanı sürücüsünü temsil eden bir driver parametresi alır. self.driver özniteliği, sağlanan driver değerine ayarlanır. -
Bu connect_to_database fonksiyonu, @abstractmethod dekoratörü kullanılarak soyut bir fonksiyon olarak işaretlenir. Veritabanının alt sınıfları, belirli bir veritabanıyla bağlantı kurması gereken bu fonksiyonu implement etmelidir.
-
__enter__
fonksiyonu, Python'un context manager protokolünde kullanılan özel bir fonksiyondur. Context manager'a girilirken çağrılır (with deyimi kullanılarak). Burada connect_to_database fonksiyonunu çağırarak bir veritabanı bağlantısı kurar ve bağlantıyı self.connection'a atar. Ayrıca, SQL sorgularını yürütmek için bir cursor nesnesi (self.cursor) oluşturur. -
__enter__
fonksiyonu, with bloğu içinde bir context manager olarak Database sınıfı objesini kullanmanıza izin veren self'i döndürür. -
__exit__
fonkisyonu, context manager protokolünde kullanılan başka bir özel fonksiyondur. Context manager'dan çıkarken (with bloğundan çıkarken) çağrılır. Burada self.cursor ve veritabanı bağlantısını (self.connection) kapatır. -
__exit__
fonksiyonu üç bağımsız değişken alır: exception_type, exc_val ve traceback. Bunlar, context manager içinde meydana gelen exceptionları işlemek için kullanılır. Ancak, kodumuzda uygulanan açık bir exception işleme yoktur. -
Özetlemek gerekirse, bu kod, bir veritabanıyla çalışmak için context manager görevi gören bir Database sınıfını tanımlar. Alt sınıflarda bir connect_to_database fonksiyonunun implemente edilmesini zorunlu kılar ve bir veritabanı bağlantısı ve cursor için gerekli kurulum ve temizleme mantığını sağlar.
PostgreSQL ile çalışmak için psycopg2 paketini kurun
- PostgreSQL'i yüklediğinizi ve t_news adında bir veritabanı oluşturduğunuzu varsayıyorum.Şimdi app/database.py dosyasını düzenlemeye devam edelim
# önceki importlar
import psycopg2
# önceki kodlar ...
class PgDatabase(Database):
"""PostgreSQL Database context manager"""
def __init__(self) -> None:
self.driver = psycopg2
super().__init__(self.driver)
def connect_to_database(self):
return self.driver.connect(
host=os.getenv("DB_HOST"),
port=os.getenv("DB_PORT"),
user=os.getenv("DB_USERNAME"),
password=os.getenv("DB_PASSWORD"),
database=os.getenv("DB_NAME")
)
-
PgDatabase sınıfı, Database sınıfının bir alt sınıfı olarak tanımlanır. Özellikle bir PostgreSQL veritabanı context manager'ı temsil eder.
-
init fonksiyonu, PgDatabase nesnesini başlatır. self.driver özniteliğini PostgreSQL driver kütüphanesi olan psycopg2 olarak ayarlar. Ardından, super().init(self.driver) kullanarak üst Database sınıfının init fonksiyonunu çağırır ve psycopg2 driver'ı argüman olarak iletir.
-
psycopg2 driver'ı kullanarak PostgreSQL veritabanına bağlantı kurmak için burada connect_to_database fonksiyonu implemente edilir. DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD ve DB_NAME gibi ortam değişkenlerinden (os.getenv) bağlantı ayrıntılarını alır. Bu ortam değişkenleri, uygun veritabanı yapılandırmasıyla ayarlanmalıdır.
-
Bağlantıyı kurmak için alınan bağlantı ayrıntılarıyla psycopg2'nin connect fonksiyonu çağrılır. Fonksiyon, connection nesnesini döndürür.
-
.env dosyasını oluşturalım ve aşağıdakileri ekleyelim (Lütfen değerleri kendi bilgilerinize göre değiştirin):
- Özetlemek gerekirse, bu kod, PostgreSQL veritabanı yönetimi için özel olarak hazırlanmış, Database sınıfının bir alt sınıfı olan PgDatabase sınıfını tanımlar. PostgreSQL sürücüsünü (psycopg2) kurar ve bağlantı ayrıntıları için sürücü ve ortam değişkenlerini kullanarak veritabanıyla bağlantı kurmak için connect_to_database fonksiyonunu implemente eder.
Tablo Oluşturma¶
app/database.py içine aşağıdaki fonksiyonu ve tablo adı değişkenini ekleyin
# önceki kodlar
t_news = "t_news"
def create_tables():
with PgDatabase() as db:
db.cursor.execute(f"""CREATE TABLE {t_news} (
id SERIAL PRIMARY KEY,
published_date TIMESTAMPTZ,
created_date TIMESTAMPTZ DEFAULT NOW(),
created_by VARCHAR(140),
context TEXT NOT NULL
);
""")
db.connection.commit()
print("Tables are created successfully...")
-
create_tables fonksiyonu, veritabanında gerekli tabloları oluşturmak için tanımlanır.
-
PgDatabase sınıfının bir objesi ile (db: olarak PgDatabase() ile) with deyimini kullanarak bir context manager'ı kullanır. Bu, veritabanı bağlantısının düzgün bir şekilde yönetilmesini ve kod bloğu yürütüldükten sonra otomatik olarak kapatılmasını sağlar.
-
with bloğunun içinde, tablo oluşturan bir SQL sorgusunu yürütmek için db.cursor nesnesinin çalıştırma(execute) fonksiyonu çağrılır. Tablo adının t_news değişkeninde saklanması beklenir (kodda tanımlandığı varsayılarak). CREATE TABLE deyimi, tablo yapısını id, published_date, created_date, created_by ve context gibi sütunlarla tanımlar.
-
CREATE TABLE sorgusu çalıştırıldıktan sonra, değişiklikler db.connection.commit() kullanılarak veritabanına kaydedilir. Bu, tablo oluşturma işleminin kalıcı olmasını sağlar.
-
Son olarak, konsola tablo(lar)ın başarıyla oluşturulduğunu belirten bir başarı mesajı print ile yazdırılır.
Tabloların Kaldırılması(Drop Tables)¶
Tablo zaten varsa, tabloları kaldırmak için database.py içine başka bir fonksiyon ekleyelim. app/database.py
def drop_tables():
with PgDatabase() as db:
db.cursor.execute(f"DROP TABLE IF EXISTS {t_news} CASCADE;")
db.connection.commit()
print("Tables are dropped...")
Endpoint Aracılığıyla Tabloları Kaldırma ve Oluşturma¶
Tabloları kaldırmak ve oluşturmak için bir uç nokta(endpoint) oluşturalım. app/main.py dosyasını açın ve aşağıdaki kodları ekleyin.
from fastapi import FastAPI, status # yeni
from fastapi.exceptions import HTTPException # yeni
# internals
from app.database import drop_tables, create_tables # yeni
app = FastAPI()
# önceki kodlar ...
@app.post('/initdb')
async def initdb():
try:
drop_tables()
create_tables()
return {"message": "Tables dropped and created!"}
except Exception as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Error {e}"
)
-
import ifadeleri, istisnaları işlemek için gerekli modülleri (fastapi.exceptions.HTTPException) ve app.database modülünden drop_tables ve create_tables fonksiyonlarını içe aktarmak için eklenir.
-
/initdb yolu, @app.post dekoratörü kullanılarak bir HTTP POST isteklerini kabul eden bir yol olarak tanımlanır. Veritabanı tablolarının kaldırılmasından ve oluşturulmasından sorumludur.
-
initdb fonksiyonunun içinde, tabloların kaldırılması ve oluşturulması sırasında meydana gelebilecek olası istisnaları(exceptions) işlemek için bir try-except bloğu kullanılır.
-
Try bloğunda, mevcut tabloları kaldırmak için drop_tables işlevi çağrılır ve ardından yeni tablolar oluşturmak için create_tables işlevi çağrılır.
-
İşlemler herhangi bir istisna olmadan tamamlanırsa, bir başarı mesajıyla birlikte bir JSON yanıtı döndürülür: {"message": "Tables dropped and created!"}.
-
İşlem sırasında bir istisna meydana gelirse, istisna bloğu tetiklenir. 400 (Bad Request) durum kodu ve belirli hatayı (f"Error {e}") içeren ayrıntılı bir hata mesajıyla bir HTTPException oluşturulur.
-
Şimdi uç noktayı test edelim. Bu uç noktaya POST isteği göndermek için curl terminali açın veya postman, thunder client istemcisini(client) kullanabilirsiniz.
- Cevap olarak aşağıdaki bilgi döner
Part 2¶
5. Ham SQL ile CRUD Uç Noktalarını Oluşturma¶
Modeller ve Şemalar Oluşturma app klasörü altında models.py dosyası oluşturun ve aşağıdaki kodları ekleyin:
from pydantic import BaseModel, Field
class NewsSchema(BaseModel):
created_by: str = Field(..., min_length=3, max_length=140)
context: str = Field(..., min_length=3, max_length=4096)
published_date: str = Field(..., min_length=3, max_length=32)
class NewsDB(NewsSchema):
id: int
-
NewsSchema sınıfı, Pydantic kütüphanesinden BaseModel'in bir alt sınıfı olarak tanımlanır. Bir news öğesinin şemasını veya yapısını temsil eder.
-
NewsSchema sınıfının içinde, Pydantic'ten Field sınıfı kullanılarak tanımlanan üç alan vardır:
-
create_by alanı string (str) türündedir ve sırasıyla 3 ve 140 karakterlik minimum ve maksimum uzunluk kısıtlamalarına sahiptir.
-
context alanı da string (str) türündedir ve sırasıyla 3 ve 4096 karakterlik minimum ve maksimum uzunluk kısıtlamalarına sahiptir.
-
published_date alanı, dize (str) türündedir ve sırasıyla 3 ve 32 karakterlik minimum ve maksimum uzunluk kısıtlamalarına sahiptir.
-
NewsDB sınıfı, NewsSchema'nın bir alt sınıfı olarak tanımlanır. Ek id alanı da dahil olmak üzere bir news öğesi için veritabanı modelini temsil eder.
-
id alanı tamsayı (int) türündedir ve veri tabanındaki bir news öğesinin benzersiz tanımlayıcısını temsil eder.
-
Bu modelleri tanımlayarak, gelen verileri doğrulamak, JSON'u serialize etmek ve veri yapısı ve türleri üzerindeki kısıtlamaları uygulamak için kullanabiliriz. database.py dosyasına SELECT sorgusu yazalım
# önceki kodlar ... def select_t_news(): with PgDatabase() as db: db.cursor.execute(f"""SELECT id, created_by, context, published_date FROM {t_news};""") objects = [ { "id": data[0], "created_by": data[1], "context":data[2], "published_date":str(data[3]) } for data in db.cursor.fetchall() ] return objects
-
select_t_news işlevi, t_news tablosundan veri almak için tanımlanır.
-
PgDatabase sınıfının bir örneğiyle (db: olarak PgDatabase() ile) with deyimini kullanarak bir context manager yöneticisi kullanır. Bu, veritabanı bağlantısının düzgün bir şekilde yönetilmesini ve kod bloğu yürütüldükten sonra otomatik olarak kapatılmasını sağlar.
-
with bloğunun içinde, t_news tablosundan veri seçen bir SQL sorgusu yürütmek için db.cursor nesnesinin yürütme yöntemi çağrılır. Seçilen sütunlar id, yaratılan_by, bağlam ve yayınlanma_tarihi'dir.
-
SQL sorgusunu yürüttükten sonra, db.cursor'da fetchall yöntemi çağrılarak seçilen tüm satırları demetler listesi olarak alır. Her demet, sonuç kümesinden bir satırı temsil eder.
-
Satırları yinelemek ve sözlüklerin (nesnelerin) bir listesini oluşturmak için bir liste kavrama kullanılır. Her sözlük, "id", "created_by", "context" ve "published_date" anahtarlarıyla bir haberi temsil eder. Değerler, demetlerdeki karşılık gelen öğelerden çıkarılır. "published_date" değeri, str() kullanılarak bir dizgeye dönüştürülür.
-
Son olarak, seçilen haber öğelerini sözlük olarak içeren nesneler listesi döndürülür.
-
database.py dosyasında SELECT by ID sorgusu yazalım
# önceki kodlar ... def select_t_news_by_id(id: int) -> dict: with PgDatabase() as db: db.cursor.execute(f""" SELECT id, created_by, context, published_date FROM {t_news} WHERE id={id}; """) data = db.cursor.fetchone() if data is None: return None return { "id": data[0], "created_by": data[1], "context": data[2], "published_date": str(data[3]) }
-
select_t_news_by_id fonksiyonu, verilen id'ye göre t_news tablosundan belirli bir haber objesini almak için tanımlandı.
-
Sorgu, verilen id ile eşleşen id sütununa göre satırları filtreleyen bir WHERE yan sorgusu içerir.
-
SQL sorgusunu yürüttükten sonra, sorguyla eşleşen ilk satırı almak için db.cursor'da fetchone fonksiyonu çağrılır. Sonuç, seçilen satırı temsil eden bir tuple olan data değişkeninde saklanır.
-
Veriler Yok ise, verilen id'ye sahip hiçbir satır bulunmadığı anlamına gelir, bu nedenle None döndürülür.
-
Veriler geçerli bir satır içeriyorsa, "id", "created_by", "context" ve "published_date" anahtarlarıyla bir sözlük(dict) oluşturulur. Değerler, tuple'da karşılık gelen öğelerden çıkarılır. "published_date" değeri, str() kullanılarak bir stringe dönüştürülür.
-
Son olarak, seçilen news objesini temsil eden dict döndürülür.
-
database.py dosyasına INSERT sorgusu yazalım
# önceki importlar ... # internals from app.models import NewsDB, NewsSchema # new # önceki kodlar ... def insert_t_news(payload: NewsSchema, *args, **kwargs) -> NewsDB: with PgDatabase() as db: db.cursor.execute(f""" INSERT INTO {t_news}(created_by, context, published_date) VALUES('{payload.created_by}', '{payload.context}', '{payload.published_date}' ) RETURNING id; """) db.connection.commit() inserted_id = db.cursor.fetchone()[0] obj = select_t_news_by_id(inserted_id) return obj
-
with bloğunun içinde, t_news tablosuna yeni bir satır ekleyen bir SQL sorgusu yürütmek için db.cursor nesnesinin execute fonksiyonu çağrılır. Sorgu, payload nesnesinden çıkarılan değerlerle VALUES yan tümcesini içerir.
-
RETURNING yan tümcesi, yeni eklenen satırın id'sini almak için kullanılır.
-
SQL sorgusu çalıştırıldıktan sonra, eklemenin kalıcı olması için db.connection.commit() kullanılarak değişiklikler veritabanına kaydedilir.
-
Girilen satırın id değerini almak için db.cursor'da fetchone yöntemi çağrılır ve inserted_id değişkeninde saklanır.
-
select_t_news_by_id fonksiyonu, eklenen news objesini almak için insert_id ile birlikte çağrılır.
-
Son olarak, eklenen news objesi döndürülür.
-
database.py dosyasına UPDATE sorgusu yazılması
# önceki kodlar def update_t_news_by_id(id: int, payload: NewsSchema): with PgDatabase() as db: db.cursor.execute(f""" UPDATE {t_news} SET created_by='{payload.created_by}', context='{payload.context}', published_date='{payload.published_date}' WHERE id='{id}' RETURNING id; """) db.connection.commit() result = db.cursor.fetchone() if not result: return None updated_id = result[0] obj = select_t_news_by_id(updated_id) return obj
-
database.py dosyasına DELETE sorgusu yazalım
# önceki kodlar...
def delete_t_news_by_id(id: int):
with PgDatabase() as db:
db.cursor.execute(f"""
DELETE FROM {t_news}
WHERE id={id};
""")
db.connection.commit()
res = db.cursor.statusmessage
if res == "DELETE 1":
return True
return False
-
Silme işleminin sonucunu belirlemek için db.cursor nesnesinin statusmessage niteliği kontrol edilir. Durum mesajı "DELETE 1" ise, bu, bir satırın başarıyla silindiği anlamına gelir, bu nedenle True döndürülür.
-
Durum mesajı "DELETE 1" değilse, bu, hiçbir satırın silinmediği veya bir hata oluştuğu anlamına gelir, bu nedenle False döndürülür.
-
CRUD view fonksiyonları yazalım
-
app klasörü altında views.py dosyası oluşturun
-
create_news fonksiyonu
from fastapi import APIRouter, HTTPException, status from psycopg2.errors import DatetimeFieldOverflow # internals from app.database import (insert_t_news,) from app.models import NewsDB, NewsSchema router = APIRouter() @router.post('/', response_model=NewsDB, status_code=status.HTTP_201_CREATED) async def create_news(payload: NewsSchema): try: res = insert_t_news(payload) return res except DatetimeFieldOverflow: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Formats are : month-day-year hour:minute:seconds or year-month-day hour:minute:seconds" )
-
router oluşturmak için FastAPI'den APIRouter sınıfı içe aktarılır.
-
İstisnaları(Exceptions) işlemek için FastAPI'den HTTPException sınıfı içe aktarılır.
-
psycopg2.errors'tan DatetimeFieldOverflow istisna sınıfı içe aktarılır. Tarih-saat alanı yanlış formatlarla ilgili istisnaları özellikle işler.
-
app.database modülünden insert_t_news işlevi içe aktarılır. Veritabanına bir news objesi eklemek için kullanılır.
-
app.models modülünden NewsDB ve NewsSchema modelleri içe aktarılır. Sırasıyla response modeli ve payload modeli olarak kullanılırlar.
-
Bir APIRouter objesi oluşturulur ve router değişkenine atanır.
-
@router.post dekoratörü, bir news objesi oluşturmak için POST isteklerini kabul etmek için kullanılır. Endpoint(uç nokta, path) olarak kök yolu ("/") belirtir.
-
response_model parametresi, bu rota için yanıt modelini temsil eden NewsDB olarak ayarlanır. Yanıtın NewsDB modeline göre serialize edilmesini sağlar.
-
status_code parametresi, yanıt durum kodunun 201 Oluşturulacağını belirten status.HTTP_201_CREATED olarak ayarlanır.
-
view fonksiyonu, create_news adıyla tanımlanır. İstek gövdesinde gönderilen verileri temsil eden NewsSchema türünde bir payload parametresi alır.
-
Fonksiyon içinde, insert_t_news(payload) çağrılarak haber veri tabanına eklenmeye çalışılır. Sonuç, res değişkeninde saklanır.
-
Ekleme başarılı olursa, yanıt olarak res değişkeni döndürülür.
-
Ekleme sırasında bir DatetimeFieldOverflow istisnası oluşursa bu, sağlanan tarih saat değerinin alanın kapasitesini aştığı anlamına gelir. Bu durumda, 400 Hatalı İstek durum kodu ve desteklenen tarih/saat biçimlerini gösteren bir ayrıntı mesajı ile bir HTTPException oluşturulur.
-
Kod, uygun yanıt modelleri ve tarih/saat alan taşmaları için hata işleme ile news objelerinin oluşturulmasını işlemek için bir API yolu(path) oluşturur.
-
/news uç noktalarını aşağıdaki gibi app/main.py dosyasına dahil etmemiz gerekiyor
from fastapi import FastAPI, status from fastapi.exceptions import HTTPException # internals from app.database import drop_tables, create_tables from app import views # yeni app = FastAPI() # önceki kodlar... # sayfanın en sonuna eklenir app.include_router(views.router, prefix='/news', tags=['news'])
- create_news'i test etmek için request body ile birlikte http://0.0.0.0:8000/news/ adresine POST isteği yapın
- Cevap aşağıdaki gibi olacaktır
-
read_news fonksiyonu
from typing import List # yeni from fastapi import APIRouter, HTTPException, status from psycopg2.errors import DatetimeFieldOverflow, OperationalError # yeni # internals from app.database import (insert_t_news, select_t_news) # yeni from app.models import NewsDB, NewsSchema # önceki kodlar... @router.get('/', response_model=List[NewsDB], status_code=status.HTTP_200_OK) async def read_news(): try: return select_t_news() except OperationalError: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="""Check if the database exists, connection is successful or tables exist. To create tables use '/initdb' endpoint""" ) except Exception as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"""Error {e}""" )
-
@router.get dekoratörü, news objelerini almak için bir GET rotası tanımlamak için kullanılır. Kök yol ("/") endpoint olarak kullanılır.
-
response_model parametresi, Yanıtın NewsDB modeline göre serialize edilmiş news objelerinin bir listesi olacağını belirten List[NewsDB] olarak ayarlanmıştır.
-
status_code parametresi, status.HTTP_200_OK olarak ayarlanır ve yanıt durum kodunun 200 OK olacağını belirtir.
-
Rota işlevi read_news adıyla tanımlanır ve parametre içermez.
-
Fonksiyonun içinde, select_t_news fonksiyonu çağrılarak tüm news objeleri alınmaya çalışılır.
-
Alma başarılı olursa, alınan news objeleri yanıt olarak döndürülür.
-
Alma sırasında bir OperationalError istisnası oluşursa, bu, veritabanıyla ilgili sunucu tarafında bir sorun olduğunu gösterir. Bu durumda, 500 Dahili Sunucu Hatası durum kodu ve veritabanının varlığının, başarılı bağlantısının veya tablo varlığının kontrol edilmesini öneren bir ayrıntı mesajıyla bir HTTPException oluşturulur. Ayrıca tablo oluşturmak için /initdb uç noktasından da bahseder.
-
Alma sırasında başka bir özel durum oluşursa, 400 Hatalı İstek durum koduyla genel bir HTTPException oluşturulur. İstisna ayrıntıları, yanıtın ayrıntı mesajına dahil edilir.
-
Bu kod, uygun response modelleri ve operasyonel hatalar ve diğer istisnalar için hata işleme ile tüm haber öğelerini almak için bir GET yolu kurar.
-
read_news'i test etmek için http://0.0.0.0:8000/news/ adresine GET isteği yapın
-
Şunun gibi bir yanıt alacaksınız: (Biz 2 tane daha veri eklemiştik)
[ { "created_by": "adnankaya", "context": "fastapi tutorial by adnankaya", "published_date": "2023-07-14 14:53:21+03:00", "id": 1 }, { "created_by": "adnankaya", "context": "django tutorial by adnankaya", "published_date": "2023-06-14 14:53:21+03:00", "id": 2 }, { "created_by": "adnankaya", "context": "flask tutorial by adnankaya", "published_date": "2022-11-22 14:53:21+03:00", "id": 3 } ]
-
read_news_by_id fonksiyonu
from typing import List from fastapi import APIRouter, HTTPException, status, Path # new from psycopg2.errors import DatetimeFieldOverflow, OperationalError # internals from app.database import (insert_t_news, select_t_news, select_t_news_by_id) # new from app.models import NewsDB, NewsSchema @router.get('/{id}/', response_model=NewsDB, status_code=status.HTTP_200_OK) async def read_news_by_id(id: int = Path(..., gt=0)): result = select_t_news_by_id(id) if not result: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='News not found') return result
-
@router.get dekoratörü, bir news objesinii id'ye göre almak için bir GET rotası tanımlamak için kullanılır. /{id}/ path parametresi, rotanın URL'sinde belirtilmiştir.
-
response_model parametresi, yanıtın NewsDB modeline göre serialize edilmiş bir news objesi olacağını belirten NewsDB olarak ayarlanır.
-
status_code parametresi, status.HTTP_200_OK olarak ayarlanır ve yanıt durum kodunun 200 OK olacağını belirtir.
-
Route işlevi read_news_by_id adıyla tanımlanır ve int türünde bir id parametresi alır. Path sınıfı, parametreyi URL'de bir yol parametresi olarak tanımlamak için fastapi.Path'ten kullanılır.
-
id parametresi, Path yapıcısında ... ve gt=0 kullanılarak doğrulanır; bu, id'nin sağlanması gerektiğini ve 0'dan büyük olduğunu gösterir.
-
Fonksiyon içinde, sağlanan id'ye göre haber öğesini almak için select_t_news_by_id işlevi çağrılır. Sonuç, result değişkeninde saklanır.
-
Sonuç, sağlanan id'ye sahip hiçbir news objesinin bulunmadığını belirten Yok ise, 404 Bulunamadı durum kodu ve haber öğesinin bulunamadığını belirten bir ayrıntı mesajı ile bir HTTPException oluşturulur.
-
Sonuç None değilse, news objesi yanıt olarak döndürülür.
-
Bu kod, haberin bulunamadığında durum için uygun yanıt modelleri ve hata işleme ile kimliğine göre belirli bir haber öğesini almak için bir GET rotası kurar.
-
read_news_by_id test etmek için http://0.0.0.0:8000/news/1 adresine GET isteği yapın
-
Şunun gibi bir yanıt alacaksınız:
-
update_news_by_id fonksiyonu
# internals from app.database import (insert_t_news, select_t_news, select_t_news_by_id, update_t_news_by_id) # yeni @router.put('/{id}/', response_model=NewsDB, status_code=status.HTTP_200_OK) async def update_news_by_id(payload: NewsSchema, id: int = Path(..., gt=0)): result = update_t_news_by_id(id, payload) if not result: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='News not found') return result
-
@router.put dekoratörü, bir news objesini id'ye göre güncellemek için bir PUT rotası tanımlamak için kullanılır. /{id}/ path parametresi, rotanın URL'sinde belirtilmiştir.
-
response_model parametresi, yanıtın NewsDB modeline göre serialize edilmiş bir news objesi olacağını belirten NewsDB olarak ayarlanır.
-
status_code parametresi, status.HTTP_200_OK olarak ayarlanır ve yanıt durum kodunun 200 OK olacağını belirtir.
-
Rota işlevi update_news_by_id adıyla tanımlanır. İstek gövdesinde gönderilen verileri temsil eden NewsSchema türünde bir payload parametresi ve haber öğesinin id'sini temsil eden int türünde bir id parametresi alır. Doğrulandıklarından ve 0'dan büyük olduklarından emin olmak için her iki parametre de ... ve gt=0 ile Path kullanılarak doğrulanır.
-
Fonksiyonun içinde, sağlanan id ve payload'a dayalı olarak veritabanındaki news objelerini güncellemek için update_t_news_by_id fonksiyonu çağrılır. Sonuç, result değişkeninde saklanır.
-
Sonuç, sağlanan id'ye sahip hiçbir news objesinin bulunamadığını veya güncellemenin başarısız olduğunu belirten Yok ise, 404 Bulunamadı durum kodu ve news objesinin bulunamadığını belirten bir ayrıntı mesajı ile bir HTTPException oluşturulur.
-
Sonuç None değilse, yanıt olarak güncellenen news objesi döndürülür.
-
Bu kod, news objesinin bulunamadığı veya güncellemenin başarısız olduğu durum için uygun yanıt modelleri ve hata işleme ile belirli bir news objesini id'ye göre güncellemek için bir PUT yolu kurar.
-
Fonksiyonu test etmek için request body birlikte http://0.0.0.0:8000/news/1 adresine PUT isteği yapın. Request body:
- Yanıt olarak aşağıdaki sonuç döner
-
delete_news_by_id fonksiyonu
from app.database import (insert_t_news, select_t_news, select_t_news_by_id, update_t_news_by_id, delete_t_news_by_id) # yeni @router.delete('/{id}/', status_code=status.HTTP_204_NO_CONTENT) async def delete_news_by_id(id: int = Path(..., gt=0)): result = delete_t_news_by_id(id) if not result: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='News not found') return result
-
@router.delete dekoratörü, bir news objesini id'ye göre silmek için bir DELETE yolu tanımlamak için kullanılır. /{id}/ path parametresi, rotanın URL'sinde belirtilmiştir.
-
status_code parametresi, status.HTTP_204_NO_CONTENT olarak ayarlanır ve yanıt durum kodunun 204 İçerik Yok olacağını belirtir.
-
Route işlevi, delete_news_by_id adıyla tanımlanır ve int türünde bir id parametresi alır. Path sınıfı, id parametresini URL'de bir yol parametresi olarak tanımlamak için kullanılır ve ... ve gt=0 doğrulama için kullanılır, id'nin sağlanması ve 0'dan büyük olması gerekir.
-
Fonksiyonun içinde, verilen id'ye göre news objesini veri tabanından silmek için delete_t_news_by_id fonksiyonu çağrılır. Sonuç, result değişkeninde saklanır.
-
Sonuç, sağlanan id'ye sahip hiçbir news objesinin bulunamadığını veya silme işleminin başarısız olduğunu belirten False ise, 404 Bulunamadı durum kodu ve haber öğesinin bulunamadığını belirten bir ayrıntı mesajı ile bir HTTPException oluşturulur.
-
Sonuç, silme işleminin başarılı olduğunu gösteren True ise, yanıt olarak result döndürülür. Ancak yanıt durum kodu 204 İçerik Yok olduğu için yanıt gövdesi boş olacaktır.
-
Bu kod, belirli bir news objesini id'ye göre silmek için uygun yanıt durum kodu ve news objesi bulunamadığında veya silme işlemi başarısız olduğunda durum için hata işleme ile bir DELETE yolu kurar.
-
http://0.0.0.0:8000/news/2 adresine DELETE isteği yaparak fonksiyonu test edin. Yanıt boş ve Durum: 204 İçerik Yok (HTTP 204 NO CONTENT)
Part 3¶
- Bu yazımızda pytest kullanarak otomatik testler yazacağız.
pytest ve httpx'i kurun
-
tests/init.py ve tests/test_views.py dosyaları oluşturun ve test_views.py dosyasını açın ve gerekli importlar için aşağıdaki kodları ekleyin.
-
PostgreSQL'de db_test_news veritabanı oluşturmayı unutmayın
import os import json from fastapi import status from fastapi.testclient import TestClient from datetime import datetime as dt import pytest # internals from app.main import app from app.database import create_tables, insert_t_news, drop_tables from app import models client = TestClient(app) # use test database and create tables os.environ["DB_NAME"] = "db_test_news" drop_tables() create_tables()
-
Ortam değişkeni manipülasyonu için os, JSON verileriyle çalışmak için json, HTTP durum kodları için fastapi'den durum, FastAPI uygulamasını test etmek için fastapi.testclient'ten TestClient, tarih ve saat değerleri ile çalışmak için datetime ve pytest dahil olmak üzere gerekli içe aktarmalar yapılır.
-
app.main modülündeki app objesi içe aktarılır. Bu, test için kullanılacak FastAPI app nesnesidir.
-
app.database modülünden create_tables, insert_t_news ve drop_tables fonksiyonları içe aktarılır. Bu fonksiyonlar, veritabanında tablo oluşturmak ve kaldırmak için kullanılır.
-
app paketindeki models modülü içe aktarılır. Bu modül muhtemelen uygulamada kullanılan veri modellerini içerir.
-
app nesnesiyle bir TestClient objesi oluşturulur ve client değişkenine atanır. Bu client, uygulamaya test istekleri yapmak için kullanılacaktır.
-
DB_NAME ortam değişkeni "db_test_news" olarak ayarlandı. Bu muhtemelen kullanılacak test veritabanının adını belirtecektir.
-
drop_tables fonksiyonu, test veritabanındaki tabloları kaldırmak için çağrılır.
-
create_tables fonksiyonu, test veritabanındaki tabloları oluşturmak için çağrılır.
-
Bu kod, test veritabanı tablolarının oluşturulması da dahil olmak üzere FastAPI uygulamasının test edilmesi için gerekli test ortamını kurar. TestClient örneği, uygulamaya test istekleri yapmaya hazır.
-
test_read_news işlevi, GET /news uç noktası için bir test durumu olarak tanımlanır.
-
client.get fonksiyonu, /news uç noktasına bir GET isteği yapmak için kullanılır.
-
response nesnesi, sunucudan alınan yanıtı temsil eder.
-
Assert ifadesi, yanıt durum kodunun 200 OK (status.HTTP_200_OK) olduğunu doğrulamak için kullanılır. Onaylama geçerse, test senaryosu başarılı kabul edilir. Onaylama başarısız olursa, bir onaylama hatası ortaya çıkar.
-
Bu test durumu, /news endpoint'inin 200 OK durum koduyla başarılı bir yanıt döndürdüğünü doğrular. Gerekirse response gövdesini veya yanıtın diğer yönlerini doğrulamak için daha fazla test senaryosu ekleyebilirsiniz.
def test_create_news(): payload = { 'created_by': 'adnankaya', 'context': 'contex1', 'published_date': '05-23-2023 15:15:19' } expected = { 'id': 1, 'created_by': 'adnankaya', 'context': 'contex1', 'published_date': '2023-05-23 15:15:19+03:00' } response = client.post('/news/', data=json.dumps(payload)) assert response.status_code == status.HTTP_201_CREATED res_json = response.json() assert res_json["created_by"] == expected["created_by"] assert res_json["context"] == expected["context"] req = dt.strptime(res_json["published_date"], '%Y-%m-%d %H:%M:%S%z') res = dt.strptime(expected["published_date"], '%Y-%m-%d %H:%M:%S%z') assert req == res response = client.post('/news/', json={'created_by': 'cb', 'context': 'ct'}) assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
- Bu test durumu, bir news objesi oluşturmak için POST /news/ uç noktasının davranışını doğrular. Yanıt durum kodunu, yanıt verilerini kontrol eder ve gerekli alanların eksik olduğu durumları ele alır.
- Bu test durumu, gerekli alanları eksik olan geçersiz bir news objesi oluşturulduğunda, sunucunun 422 durum koduyla doğru bir şekilde yanıt döndürdüğünü doğrular.
def test_read_news_by_id(): payload = models.NewsSchema(created_by='single_object', context='context single_object', published_date='01-02-2023 13:14:15') # insert database obj = insert_t_news(payload) # retrieve response = client.get(f'/news/{obj["id"]}') assert response.status_code == 200 assert response.json()['id'] == obj['id'] assert response.json()['created_by'] == obj['created_by'] assert response.json()['context'] == obj['context']
-
Bu test durumu, GET /news/{id} uç noktasının id'ye göre doğru news objesini aldığını ve yanıtın beklenen verileri içerdiğini doğrular.
-
Bu kod, GET /news/{id} uç noktası aracılığıyla bir news objesini almaya çalışırken geçersiz bir id'nin sağlandığı durumu test etmek için test_invalid_read_news_by_id adlı bir test işlevini tanımlar.
-
Bu test durumu, geçersiz bir id sağlandığında, sunucunun var olmayan bir kimlik için 'News not found' ayrıntı mesajıyla birlikte 404 durum kodunu ve geçersiz bir id değeri için 422 durum kodunu doğru bir şekilde döndürdüğünü doğrular.
def test_update_news(): payload = {"created_by": "adnankaya_updated", "context": "context1_updated", "published_date": "01-22-2071 19:18:17"} response = client.put("/news/1/", data=json.dumps(payload)) assert response.status_code == status.HTTP_200_OK res_json = response.json() assert res_json["created_by"] == payload["created_by"] assert res_json["context"] == payload["context"] req = dt.strptime(res_json["published_date"], '%Y-%m-%d %H:%M:%S%z') res = dt.strptime(payload["published_date"], '%m-%d-%Y %H:%M:%S') assert req.year == res.year assert req.month == res.month assert req.day == res.day assert req.hour == res.hour assert req.minute == res.minute assert req.second == res.second
-
Bu kod, bir news objesini id'ye göre güncellemek için PUT /news/{id} uç noktasını test etmek üzere test_update_news adlı bir test fonksiyonu tanımlar. Bu test senaryosu, PUT /news/{id} uç noktasının news objesini sağlanan verilerle doğru bir şekilde güncellediğini doğrular ve güncellenen news objesini beklenen değerlerle döndürür.
@pytest.mark.parametrize( 'id, payload, status_code', [ [1, {}, 422], [1, {'context': 'var'}, 422], [44, {'created_by': 'adnan', 'context': 'context', 'published_date': '12-19-2022 13:14:15'}, 404], [1, {'created_by': 'c', 'context': 'var'}, 422], [1, {'created_by': 'adnan', 'context': 'd'}, 422], [0, {'created_by': 'adnan', 'context': 'context'}, 422], ], ) def test_update_invalid_news(id, payload, status_code): response = client.put(f'/news/{id}/', data=json.dumps(payload)) assert response.status_code == status_code
-
Bu kod, PUT /news/{id} uç noktası yoluyla geçersiz news objelerini güncellemenin çeşitli durumlarını test etmek için test_update_invalid_news adlı parametreleştirilmiş bir test işlevini tanımlar.
-
Bu parametreleştirilmiş test durumu, gerekli alanların eksik veya geçersiz olduğu durumlar da dahil olmak üzere, geçersiz news objelerinin güncellenmesine yönelik birden çok senaryonun test edilmesine olanak tanır.
-
Her test durumu, farklı senaryoları kapsayacak şekilde belirli bir id, payload ve beklenen durum_kodu sağlar.
def test_delete_news_by_id(): payload = models.NewsSchema(created_by='deleteable_object', context='context deleteable_object', published_date='01-02-2023 13:14:15') # insert database obj = insert_t_news(payload) # delete request response = client.delete(f'/news/{obj["id"]}/') assert response.status_code == status.HTTP_204_NO_CONTENT assert response.text == '' response = client.delete('/news/0/') assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
- Bu kod, bir news objesini id'ye göre silmek için DELETE /news/{id} uç noktasını test etmek üzere test_delete_news_by_id adlı bir test işlevi tanımlar.
- Bu test durumu, DELETE /news/{id} uç noktasının, sağlanan id'ye sahip news objesini doğru bir şekilde sildiğini ve 204 NO CONTENT durum kodu döndürdüğünü doğrular.
- Ayrıca, geçersiz bir id'nin sağlandığı durumu da ele alır ve 422 UNPROCESSABLE ENTITY durum kodu bekler.
- Bu kod, DELETE /news/{id} uç noktası aracılığıyla bir news objesini silmeye çalışırken geçersiz bir id'nin sağlandığı durumu test etmek için test_invalid_delete_news_by_id adlı bir test işlevi tanımlar.
- Bu test durumu, geçersiz bir id sağlandığında, sunucunun var olmayan bir id için 'Haber bulunamadı' ayrıntı mesajıyla birlikte doğru bir şekilde 404 durum kodu döndürdüğünü doğrular.
Testleri Çalıştıralım¶
- diyerek çalıştırabilirsiniz. Eğer herhangi bir hata yoksa aşağıdaki gibi bir çıktı almanız beklenir.=================== test session starts ====================
platform darwin -- Python 3.11.0, pytest-7.4.0, pluggy-1.2.0
rootdir: /Users/dev/webdev/tutorial/news_fastapi
plugins: anyio-3.7.1
collecting ... Tables are dropped...
Tables are created successfully...
collected 14 items
tests/test_views.py ..............
================= 14 passed, 9 warnings in 0.19s ===========
-
Bu yazı ile beraber serinin sonuna gelmiş bulunuyoruz.
-
Test yazalım yazdıralım :)
-
Kodlara erişmek için github repo : https://github.com/adnankaya/news_fastapi