mssql-python'u FastAPI ile kullanın

FastAPI, API oluşturmak için modern bir Python web çerçevesidir. mssql-python ile birleştiğinde, Microsoft SQL ve Azure SQL Veritabanı destekli yüksek performanslı REST API'leri oluşturabilirsiniz.

Prerequisites

  • Python 3.10 veya üzeri.
  • mssql-python, fastapi, uvicorn, pydantic ve PyJWT paketleri. Tümünü pip install fastapi uvicorn mssql-python pydantic pyjwt ile yükleyin.
  • Tek seferlik işletim sistemine özgü önkoşulları yükleyin. Windows kullanıcıları bu adımı atlayabilir. Platformun tam detayları için mssql-python'u Install sayfasına bakınız.
    apk add libtool krb5-libs krb5-dev

SQL veritabanı oluşturma

Aşağıdaki platformlardan birinde bir SQL veritabanı oluşturun veya bağlanın:

Bu makaledeki örnekler AdventureWorksLT örnek veritabanını, özellikle SalesLT.Product tabloyu kullanır. Eğer AdventureWorksLT'niz yüklü değilse, AdventureWorks örnek veritabanlarına bakabilirsiniz.

Proje kurulumu

Sanal ortam oluşturma

Bu projenin paketlerinin diğer Python kurulumlarından izole kalması için sanal bir ortam oluşturun ve etkinleştirin. Bu adım, uygulamanızı çalıştırırken paketleri bir yorumlayıcıya yükleme veya başka bir uygulamada test yapma sorununu da önler.

py -m venv .venv
.\.venv\Scripts\Activate.ps1

Ortamı etkinleştirdikten sonra, python, pip ve pytest üçü de aynı yorumlayıcıyı işaret eder. Bu makaledeki kalan komutları etkinleştirilmiş ortamdan çalıştırın.

Note

Windows on Arm’de, ortamı Python’ın Arm64 derlemesiyle oluşturun; böylece mssql-python ve bağımlılıkları önceden oluşturulmuş wheel paketlerinden yüklenir. Birden fazla Python sürümü bulunan bir makinede, py -m venv beklediğinizden farklı bir sürümü veya mimariyi seçebilir; bu yüzden etkinleştirdikten sonra python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" ile doğrulayın. Eğer pip, cryptography paketini kaynaktan derlemeye çalışırsa (bu bir Rust ve OpenSSL araç zinciri hatasıdır), önce pip install --only-binary=:all: cryptography ile wheel tabanlı bir sürüm yükleyin, ardından geri kalanını yükleyin.

Bağımlılıkları yükleme

Gerekli paketleri pip ile kurun:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Proje yapısı

Projenizi veritabanı, şemalar ve CRUD işlemleri için ayrı modüllerle organize edin:

my_api/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── test_api.py
└── routers/
    └── products.py

Veritabanı bağlantı yönetimi

FastAPI, yönlendirme işleyicileri için veritabanı bağlantıları gibi kaynaklar sağlamak için bağımlılık enjeksiyonu kullanır. Bu bölümdeki desen, bir bağlantı açan ve imleç veren ve otomatik olarak commit/rollback/close işlemlerini yöneten bir bağlam yöneticisi oluşturur.

database.py oluştur

Fonksiyon, get_connection_string() yapılandırma değerlerinden ODBC bağlantı dizesi'i oluşturur. get_db() bağlam yöneticisi ile get_db_dependency() üreteci aynı kalıbı izler: bir bağlantı açar, bir imleç döndürür, başarılı olursa değişiklikleri kaydeder, hata durumunda geri alır ve iş bitince her zaman kapatır. FastAPI Depends() her istek için bir kez çağrı get_db_dependency() yapar ve yaşam döngüsünü yönetir.

# database.py
import mssql_python
from contextlib import contextmanager
from typing import Generator

# Configuration
DATABASE_CONFIG = {
    "server": "<server>.database.windows.net",
    "database": "<database>",
}

def get_connection_string() -> str:
    """Build connection string from config."""
    return (
        f"Server={DATABASE_CONFIG['server']};"
        f"Database={DATABASE_CONFIG['database']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Note

ActiveDirectoryDefault DefaultAzureCredential, kullanır ve bu da birden fazla kimlik doğrulama sağlayıcısını sırayla dener. İlk bağlantı yavaş olabilir çünkü SDK çalışan bir sağlayıcı bulana kadar zincirde yürür. Üretimde, ortamınızın hangi kimlik bilgisi türünü kullandığını biliyorsanız, zincirde dolaşmayı önlemek için bunu doğrudan belirtin (örneğin, yönetilen kimlik için ActiveDirectoryMSI). Daha fazla bilgi için bkz . Microsoft Entra kimlik doğrulaması.

@contextmanager
def get_db() -> Generator:
    """Database connection context manager for FastAPI dependency injection."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

def get_db_dependency():
    """FastAPI dependency for database cursor."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

Pydantik modeller

Pydantik modeller, talep ve yanıt verileri için şekil ve doğrulama kurallarını tanımlar. FastAPI, gelen JSON'u ayrıştırmak, alan kısıtlamalarını doğrulamak ve OpenAPI dokümantasyonunu otomatik olarak oluşturmak için bu modelleri kullanır.

schemas.py oluştur

Şemaları , Base, Create, ve yanıt varyantlarına ayırınUpdate. Şema Base paylaşılan alanları tutar, Create ekleme işlemleri için ondan miras alır ve Update kısmi güncellemeler için tüm alanları isteğe bağlı hale getirir.

# schemas.py
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from typing import Optional
from datetime import datetime

# Product schemas
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    product_number: str = Field(..., min_length=1, max_length=25)
    price: float = Field(..., gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class ProductCreate(ProductBase):
    pass

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    product_number: Optional[str] = Field(None, min_length=1, max_length=25)
    price: Optional[float] = Field(None, gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class Product(ProductBase):
    id: int

    model_config = ConfigDict(from_attributes=True)

# Pagination
class PaginatedResponse(BaseModel):
    items: list
    total: int
    page: int
    page_size: int
    pages: int

CRUD operasyonları

Rota yöneticilerini ince tutmak için veritabanı sorgularını özel bir sınıfa kapsülleyin. Her statik yöntem, FastAPI tarafından enjekte edilen bir imleç alır ve SQL enjeksiyonunu önlemek için parametreli sorgular (%(name)s değer sözlüğüne sahip yer tutucular) kullanarak bir işlem uygular. Bu ayrım, iş mantığını test etmeyi ve yeniden kullanmayı kolaylaştırır.

crud.py oluşturun

# crud.py
from typing import Optional, List
from schemas import ProductCreate, ProductUpdate, Product

class ProductCRUD:
    """CRUD operations for products."""
    
    @staticmethod
    def get(cursor, product_id: int) -> Optional[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE ProductID = %(id)s
        """, {"id": product_id})
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def get_all(cursor, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]
    
    @staticmethod
    def count(cursor) -> int:
        cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
        return cursor.fetchval()
    
    @staticmethod
    def create(cursor, product: ProductCreate) -> dict:
        cursor.execute("""
            INSERT INTO SalesLT.Product (Name, ProductNumber, ListPrice, Color, Size, ProductCategoryID, StandardCost, SellStartDate)
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
        """, {
            "name": product.name,
            "product_number": product.product_number,
            "price": product.price,
            "color": product.color,
            "size": product.size,
            "category_id": product.category_id
        })
        
        row = cursor.fetchone()
        return {
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        }
    
    @staticmethod
    def update(cursor, product_id: int, product: ProductUpdate) -> Optional[dict]:
        # Build dynamic update
        updates = []
        params = {"id": product_id}
        
        if product.name is not None:
            updates.append("Name = %(name)s")
            params["name"] = product.name
        if product.product_number is not None:
            updates.append("ProductNumber = %(product_number)s")
            params["product_number"] = product.product_number
        if product.price is not None:
            updates.append("ListPrice = %(price)s")
            params["price"] = product.price
        if product.category_id is not None:
            updates.append("ProductCategoryID = %(category_id)s")
            params["category_id"] = product.category_id
        
        if not updates:
            return ProductCRUD.get(cursor, product_id)
        
        cursor.execute(f"""
            UPDATE SalesLT.Product SET {', '.join(updates)}
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            WHERE ProductID = %(id)s
        """, params)
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def delete(cursor, product_id: int) -> bool:
        cursor.execute("""
            DELETE FROM SalesLT.Product WHERE ProductID = %(id)s
        """, {"id": product_id})
        return cursor.rowcount > 0
    
    @staticmethod
    def search(cursor, query: str, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE Name LIKE %(query)s OR ProductNumber LIKE %(query)s
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"query": f"%{query}%", "skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]

FastAPI uygulaması

main.py oluştur

Ana modül her şeyi birbirine bağlıyor. Her rota, cursor = Depends(get_db_dependency) bildiriminde bulunur; bu, FastAPI'ye jeneratörü çağırmasını, üretilen imleci işleyiciye iletmesini ve ardından temizliği yapmasını söyler. FastAPI ayrıca istek gövdelerini Pydantic şemalarınıza göre handler çalışmadan önce doğrular.

# main.py
from fastapi import FastAPI, HTTPException, Depends, Query
from typing import List
from database import get_db_dependency
from schemas import Product, ProductCreate, ProductUpdate, PaginatedResponse
from crud import ProductCRUD

app = FastAPI(
    title="Product API",
    description="REST API for products using mssql-python",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Product API", "docs": "/docs"}

@app.get("/products", response_model=PaginatedResponse)
def list_products(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """List all products with pagination."""
    skip = (page - 1) * page_size
    items = ProductCRUD.get_all(cursor, skip=skip, limit=page_size)
    total = ProductCRUD.count(cursor)
    
    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    }

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Get a specific product by ID."""
    product = ProductCRUD.get(cursor, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@app.post("/products", response_model=Product, status_code=201)
def create_product(product: ProductCreate, cursor = Depends(get_db_dependency)):
    """Create a new product."""
    return ProductCRUD.create(cursor, product)

@app.put("/products/{product_id}", response_model=Product)
def update_product(
    product_id: int,
    product: ProductUpdate,
    cursor = Depends(get_db_dependency)
):
    """Update an existing product."""
    updated = ProductCRUD.update(cursor, product_id, product)
    if not updated:
        raise HTTPException(status_code=404, detail="Product not found")
    return updated

@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Delete a product."""
    if not ProductCRUD.delete(cursor, product_id):
        raise HTTPException(status_code=404, detail="Product not found")

@app.get("/products/search/", response_model=List[Product])
def search_products(
    q: str = Query(..., min_length=1),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """Search products by name or product number."""
    skip = (page - 1) * page_size
    return ProductCRUD.search(cursor, q, skip=skip, limit=page_size)

# Health check endpoint
@app.get("/health")
def health_check(cursor = Depends(get_db_dependency)):
    """Check database connectivity."""
    try:
        cursor.execute("SELECT 1")
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Database unhealthy: {str(e)}")

Uygulamayı çalıştırma

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Hata yönetimi

FastAPI, belirli istisna türleri için küresel istisna işleyicilerini kaydetmenize olanak tanır. mssql_python.DatabaseError ve mssql_python.IntegrityErroryakaladığınızda, FastAPI genel 500 yanıtlar yerine uygun HTTP durum kodlarıyla yapılandırılmış JSON hataları döndürür.

Küresel istisna işleyicisi

Bu işleyicileri, app = FastAPI(...) satırından hemen sonra main.py öğesine ekleyin. FastAPI, bir rota o istisna türünü yükselttiğinde eşleştirme handler'ı çalıştırır, yani her rotada bir try/except blok gerekmez.

# main.py
from fastapi import Request
from fastapi.responses import JSONResponse
import mssql_python

@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(request: Request, exc: mssql_python.DatabaseError):
    """Handle database errors globally."""
    return JSONResponse(
        status_code=500,
        content={"detail": "Database error occurred", "type": "database_error"}
    )

@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(request: Request, exc: mssql_python.IntegrityError):
    """Handle integrity constraint violations."""
    error_msg = str(exc)
    
    if "UNIQUE" in error_msg:
        return JSONResponse(
            status_code=409,
            content={"detail": "Resource already exists", "type": "duplicate_error"}
        )
    elif "FOREIGN KEY" in error_msg:
        return JSONResponse(
            status_code=400,
            content={"detail": "Referenced resource not found", "type": "reference_error"}
        )
    
    return JSONResponse(
        status_code=400,
        content={"detail": "Data integrity error", "type": "integrity_error"}
    )

Note

Diğer satırların hâlâ başvurduğu bir ürünü silmek, foreign key kısıtlamasından kaynaklanan mssql_python.IntegrityError hatasına yol açar ve işleyici satırı kaldırmak yerine 400 hatası döndürür. AdventureWorksLT örneğinde, SalesLT.Product içindeki ürünlerin çoğuna SalesLT.SalesOrderDetail tarafından başvurulur; bu nedenle DELETE, tasarım gereği bunlar için başarısız olur. Başarılı bir silme işlemini test etmek için, POST /products içeren bir ürün oluşturup o ürünü silin veya önce referans veren satırları kaldırın.

Bağlantı havuzlama

Bağlantı havuzu olmadan, her istek Microsoft SQL'e bir TCP bağlantısını açıp kapatır ve bu da gecikme ekler. Bağlantı havuzu, boştaki bağlantıları yeniden kullanılmak üzere hazır tutar. mssql_python.pooling() öğesini başlangıçta bir kez çağırın. Havuzlama etkinleştirildiğinde, conn.close() in get_db_dependency() bağlantıyı aslında kapatmak yerine havuza geri döndürür.

Geliştirilmiş veritabanı modülü

Havuzlamayı, başlangıçta mssql_python.pooling() çağrısı yaparak etkinleştirin ve uygun maksimum boyut ve zaman aşımı ayarlarıyla yapılandırın:

# database.py with connection pooling
import mssql_python
from contextlib import contextmanager
import os

# Configure pool
mssql_python.pooling(max_size=20, idle_timeout=300)

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

def get_db_dependency():
    """FastAPI dependency with connection pooling."""
    conn = mssql_python.connect(DATABASE_URL)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()  # Returns to pool

Kimlik doğrulama ara yazılımı

FastAPI bağımlılıklarını zincirleyerek veritabanı erişimini kimlik doğrulama ile birleştirebilirsiniz. Aşağıdaki örnek, bir JWT taşıyıcı tokenı doğrular, AdventureWorksLT örnek veritabanında eşleşen kişi kaydını arar ve sonucu korumalı rotalara sunar.

# auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    cursor = Depends(get_db_dependency)
):
    """Validate JWT and return the matching AdventureWorksLT person."""
    try:
        token = credentials.credentials
        # Replace with a strong secret loaded from environment variables
        payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
        person_id = int(payload.get("sub"))
        
        if not person_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        cursor.execute("""
            SELECT BusinessEntityID, FirstName, LastName
            FROM Person.Person
            WHERE BusinessEntityID = %(id)s
        """, {"id": person_id})
        
        person = cursor.fetchone()
        if not person:
            raise HTTPException(status_code=401, detail="User not found")
        
        return {
            "id": person.BusinessEntityID,
            "first_name": person.FirstName,
            "last_name": person.LastName
        }
        
    except (TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token subject")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Protected endpoint
@app.get("/me")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

Testing

FastAPI, gerçek bir HTTP sunucusu başlatmadan uygulamanıza istekler gönderen bir yerleşik TestClient sistem sunarhttpx. Rotaları, durum kodlarını ve yanıt yapılarını doğrulamak için pytest ile testler yazın.

Bu bölümdeki testleri çalıştırmadan önce, test bağımlılıklarını kurun:

pip install pytest httpx

Note

Starlette'in en güncel sürümünü kullanıyorsanız veya yeni bir ortam kuruyorsanız, httpx yerine httpx2 tercih edin. Starlette'in son sürümleri, TestClient için httpx2 kullanır ve yalnızca httpx kurulu olduğunda bir kullanımdan kaldırma uyarısı verir. pip install pytest httpx2 ile yükleyin.

Test kurulumu

Rota davranışını ve yanıt şemalarını doğrulamak için kullanılan bir test dosyası TestClient oluşturun:

# test_api.py
from fastapi.testclient import TestClient
from main import app
import uuid
import pytest

client = TestClient(app)

def test_list_products():
    response = client.get("/products")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data

def test_create_product():
    suffix = uuid.uuid4().hex[:8]
    name = f"Test Product {suffix}"
    product_data = {
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "color": "Red",
        "size": "M",
        "category_id": 1
    }
    response = client.post("/products", json=product_data)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == name
    assert data["price"] == 19.99

def test_get_product_not_found():
    response = client.get("/products/99999")
    assert response.status_code == 404

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

Testleri, main.py ile aynı dizin olan proje kök dizininden pytest ile çalıştırın:

pytest

Bu testler, mocklar yerine canlı veritabanınıza karşı çalışır; bu nedenle test_create_product, SalesLT.Product içine gerçek bir satır ekler. AdventureWorksLT'de her ikisinin NameProductNumber de benzersiz kısıtlamaları vardır, bu yüzden test her koşuda her biri için benzersiz bir değer üretir. Bunun yerine bu değerleri sabit olarak kodlarsanız, önce satırı silmediğiniz sürece test ikinci çalıştırmada bir çakışma nedeniyle başarısız olur.

Dağıtım yapılandırması

Pydantic'leri BaseSettings kullanarak ortam değişkenlerinden ve .env dosyalarından yapılandırma yükleyin. Bu yaklaşım, sırları kaynak koddan uzak tutar ve ortamlar arasında geçiş yapmayı kolaylaştırır. Ayarlar paketini pip install pydantic-settings ile yükleyin.

Ortam değişkenleri

Ortam değişkenlerinden yapılandırmayı yükleyen bir ayar modülü oluşturun, böylece kodunuzun dışındaki sırlar ve dağıtıma özgü değerleri yönetebilirsiniz:

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_server: str = "<server>.database.windows.net"
    database_name: str = "<database>"
    pool_size: int = 10

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Ardından, kendi kopyasını tanımlamak yerine database.py öğesini config içinden get_connection_string içe aktaracak şekilde güncelleyin. Çoğaltılan fonksiyonu kaldırarak, uygulamanın bağlantı ayarlarını tek bir kaynaktan okumasını sağlarsınız.

# database.py
from config import get_connection_string

İlgili içerik

  • Last updated on