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

- FastAPI Kurulum

(venv) $ pip install fastapi

- Sunucu olarak çalışmak için uvicorn'u da kurun:

(venv) $ pip install "uvicorn[standard]"

- news_fastapi klasörü altında app ve tests klasörleri oluşturun. Şimdilik proje dosya ve klasör yapısı aşağıdaki gibidir.

news_fastapi
├──  requirements.txt
├── app
└── tests
└── venv

- app klasörü altında main.py oluşturun ve aşağıdaki kodları ekleyin:

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def index():
    return {'message': 'Welcome to news app!'}

- Bu kod, root path (“/”) için tanımlanan tek bir rota ile temel bir FastAPI uygulaması oluşturur. Her bir parçanın ne yaptığının bir dökümü aşağıda verilmiştir:

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 --reload --workers 1 --host 0.0.0.0 --port 8000

- 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")

- BASE_DIR, pathlib'den Path sınıfı kullanılarak tanımlanır. Geçerli dosyanın (file) yolunu çözerek ve iki seviye yukarı çıkarak (parent.parent) projenin temel dizinini temsil eder. - dotenv.load_dotenv() fonksiyonu, bağımsız değişken olarak .env dosyasının (BASE_DIR / ".env") yolu ile çağrılır. Bu fonksiyon, ortam değişkenlerini .env dosyasından yükler ve uygulamada kullanılabilir hale getirir. - Projenizin temel dizininde bir .env dosyanız olduğundan ve gerekli ortam değişkenlerini içerdiğinden emin olun. dotenv.load_dotenv() fonksiyonu, ortam değişkenlerini bu dosyadan uygulamanın ortamına yükler. Veritabanı işlemlerinde kullanmak için bir context manager temel sınıfı oluşturalım. app/database.py

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()

- Database sınıfı, Abstract Base Class anlamına gelen ABC sınıfının bir alt sınıfı olarak tanımlanır. Diğer sınıflar için temel sınıf görevi görür ve alt sınıfları tarafından uygulanması gereken soyut yöntemleri tanımlayabilir.

__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

pip install psycopg2

- 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):

export DB_HOST=localhost
export DB_PORT=5432
export DB_USERNAME=developer
export DB_PASSWORD=developer
export DB_NAME=db_news

Ö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.
```
curl -X POST http://0.0.0.0:8000/initdb
```
Cevap olarak aşağıdaki bilgi döner

{"message":"Tables dropped and created!"}

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

{
  "created_by":"adnankaya",
  "context":"fastapi tutorial by adnankaya",
  "published_date":"2023-07-14 14:53:21"
}

Cevap aşağıdaki gibi olacaktır

{
  "created_by": "adnankaya",
  "context": "fastapi tutorial by adnankaya",
  "published_date": "2023-07-14 14:53:21+03:00",
  "id": 1
}

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:

{
  "created_by": "adnankaya",
  "context": "fastapi tutorial by adnankaya",
  "published_date": "2023-07-14 14:53:21+03:00",
  "id": 1
}

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:

{
  "created_by": "adnankaya new",
  "context": "fastapi & raw SQL tutorial by adnankaya",
  "published_date": "2023-07-11 14:53:21+03:00"
}

Yanıt olarak aşağıdaki sonuç döner

{
  "created_by": "adnankaya new",
  "context": "fastapi & raw SQL tutorial by adnankaya",
  "published_date": "2023-07-11 14:53:21+03:00",
  "id": 1
}

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

pip install pytest
pip install httpx

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.
```
# imports ...
def test_read_news():
    response = client.get("/news")
    assert response.status_code == status.HTTP_200_OK
```
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ı