Aller au contenu

Base de données

Starlette n'est pas strictement lié à une implémentation de base de données particulière.

Vous pouvez l'utiliser avec un ORM asynchrone tel que GINO, ou utiliser un point de terminaison non asynchrone régulier et l'intégrer à SQLAlchemy.

Dans ce document, nous allons montrer comment l'intégration peut être effectuée pour le paquet databases, qui fournit un support de base SQLAlchemy pour une gamme de pilotes de base de données différents.

Voici un exemple complet qui inclut une définition de table, la configuration d'une instance de database.Database et plusieurs points de terminaison qui interagissent avec la base de données.

**.env **

DATABASE_URL= sqlite://test.db

app.py

Importer le contexte lib

Importation de databases
Import sqlalchemy
de starlette.applications import Starlette
depuis starlette.config import Config
from starlette.responses import JSONResponse
de starlette.routing Route d'importation


# Configuration à partir de variables environnementales ou fichier '.env'.
config = Config('.env')
DATABASE_URL = config('DATABASE_URL')


#Définitions de table de base.
metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    « notes »,
    Metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("complet", sqlalchemy.Boolean),
)

databases.Database(DATABASE_URL)

@contextlib.asynccontextmanager
async def lifespan(app):
    await database.connect()
    Yield est un
    await database.disconnect()

#Code d'application principal.
async def list_notes(demande):
    query = notes.select ()
    results = await database.fetch_all(query)
    Contenu = [
        {
            "texte": résultat ["texte"],
            "Terminé": résultat ["Terminé"]
        }
        Résultats en résultats
    ]
    return JSONResponse(contenu)

async def add_note(request):
    data = await request.json()
    query = notes.insert(). valeurs ()
       text=data["text"],
       complété =data["complété"]
    )
    await database.execute(query)
    return JSONResponse({
        « texte »: données [« texte »],
        "Terminé": données ["Terminé"]
    })

Routes = [
    Route("/notes", endpoint=list_notes, methods=["GET"]),
    Route("/notes", endpoint=add_note, methods=["POST"]),
]

app = Starlette(
    routes=routes,
    Lifespan=lifespan,
)

Enfin, vous devez créer des tables de base de données. Alembic est recommandé et nous l'avons brièvement couvert dans la migration.

Questions sur

Les requêtes peuvent être effectuées en utilisant SQLAlchemy Core Query.

Les approches suivantes sont soutenues:

  • rows = await database.fetch_all(query)
  • row = await database.fetch_one(query)
  • async for row in database.iterate(query)
  • await database.execute(query)
  • await database.execute_many(query)

Transactions; affaires; transactions (ce mot a plusieurs significations, et la traduction spécifique doit être déterminée en fonction du contexte). Si la traduction littérale est faite uniquement en suivant le mot donné « Transactions », qui peut être traduit par « transaction; transaction », voici quelques paraphrases courantes pour vous aider à choisir en fonction du contexte spécifique.

Les transactions de base de données peuvent être utilisées comme décorateurs, gestionnaires de contexte ou API de bas niveau.

Utiliser un décorateur sur un point final:

@database.transaction()
async def populate_note(demande):
    #Cette base de données insère des occurrences dans une transaction.
    # It will be rolled back par le groupe `RuntimeError`.
    query = notes.insert().values(text="You won't see me", completed=True)
    await database.execute(query)
    raise RuntimeError():

Avec le gestionnaire de contexte:

async def populate_note(demande):
    async avec database.transaction():
        #Cette base de données insère des occurrences dans une transaction.
        # It will be rolled back par les `RuntimeError`.
        query = notes.insert().values(text="You won't see me", completed=True)
        await request.database.execute(query)
        raise RuntimeError()

Avec une API de bas niveau:

async def populate_note(demande):
    transaction = await database.transaction()
    Essayez:
        #Cette base de données insère des occurrences dans une transaction.
        # It will be rolled back par le `RuntimeError`
        query = notes.insert().values(text="You won't see me", completed=True)
        await database.execute(query)
        raise RuntimeError()
    Excepté:
        await transaction.rollback()
        raise
    Else:
        await transaction.commit()

Tester l'isolation

Il y a plusieurs choses que nous voulons nous assurer lorsque nous testons pour un service qui utilise une base de données.

  • Les tests ont été effectués avec une base de données distincte.
  • Une nouvelle base de données de tests est créée chaque fois que vous exécutez un test.
  • Assurez-vous que les états de la base de données entre chaque cas de test sont isolés les uns des autres.

Pour répondre à ces exigences, nous devons structurer nos applications et nos tests de cette manière:

de starlette.applications import Starlette
depuis starlette.config import Config
Importation de databases

config = Config(".env")

TESTING = config('TESTING', cast=bool, default=False)
DATABASE_URL = config('DATABASE_URL', cast=databases.DatabaseURL)
TEST_DATABASE_URL = DATABASE_URL.replace(base de données ='test_' + DATABASE_URL.database)

#Utilisez 'force_rollback' pendant le test, pour nous assurer que nous ne modifions pas la base de données persist
# between each test case.
Si vous essayez:
    databases. Database(TEST_DATABASE_URL, force_rollback=True)
Else:
    databases.Database(DATABASE_URL)

Pendant le test, nous devons toujours configurer TESTING et configurer la base de données de test. en supposant que nous utilisions py.test, voici à quoi ressemblerait notre conftest.py:

Import Pytest
de starlette.config import environ
depuis starlette.testclient import TestClient
de sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database

# This sets `os.environ`, mais fournit une protection additionnelle.
#Si nous l'avions placé sous l'importation de l'application, cela aurait causé une erreur
#nous informer que « TESTING » a déjà été lu de l'environnement.
environ ['TESTING'] = 'Vrai'

Import App


@pytest.fixture(scope="session", autouse=True)
def create_test_database():
  """
  Créez une base de données propre sur tous les cas de test.
  Pour la sécurité, nous devrions abandonner si une base de données existe déjà.

  Nous utilisons le paquet `sqlalchemy_utils` ici pour un peu d'aide en cohérence
  Créer et draper la base de données.
  """
  url = str (app. TEST_DATABASE_URL)
  engine = create_engine (url)
  assert not database_exists (url),« Test database déjà existants.
  create_database (URL)             #Créer la base de test.
  metadata. create_all (engine)      #Créez les tables.
  Yield est un                            #Exécutez les tests.
  drop_database (url)               #Drop la base de test.


@pytest. fixture ()
def client ()
    """
    Lorsque nous utilisons la fixation du client dans les cases de test, nous obtiendrons la base de données complète.
    Rollbacks entre les cases de test:

    def test_homepage (client):
        url = app. url_path_for ('home ')
        response = client. get (url)
        assert response. status_code == 200
    """
    Avec TestClient (app) comme client:
        yield client

Migration; migration

Pour gérer les modifications incrémentielles apportées à votre base de données, vous devrez presque certainement utiliser une migration de base de données-pour cela, nous recommandons fortement Alembic, qui a été écrit par les auteurs de SQLAlchemy.

$pip install alembic
$alambic init migrations

Maintenant, vous devez configurer pour qu 'Alembic fasse référence à l'URL_DATABASE configurée et utilise les métadonnées de votre table.

Supprimez la ligne suivante dans alembic.ini:

sqlalchemy.url = driver://user:pass@localhost/dbname

Dans migrations/env.py, vous devez configurer la clé de configuration 'sqlalchemy.url' et la variable target_metadata.

#L'objet de Config alambic.
config = contexte.config

#Configurer l'utilisation de notre DATABASE_URL et de nos définitions de table...
Import App
config.set_main_option('sqlalchemy.url', str(app.DATABASE_URL))
target_metadata = app.metadata

...

Ensuite, en utilisant notre exemple de note ci-dessus, créez une révision préliminaire:

Révision alambique-m "Créer une table de notes"

Remplissez le nouveau fichier (dans migrations/versions) avec les instructions nécessaires:

def upgrade():
    op. create_table(
      « notes »,
      sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
      sqlalchemy.Column("text", sqlalchemy.String),
      sqlalchemy.Column("complet", sqlalchemy.Boolean),
    )

def downgrade():
    op.drop_table('notes')

Et exécutez votre première migration. notre application de prise de notes est maintenant opérationnelle!

tête de grade supérieur alambic

Exécuter une migration pendant les tests

Chaque fois que vous créez une base de données de test, il est judicieux de vous assurer que votre suite de tests exécute une migration de base de données-cela vous aidera à détecter tout problème dans votre script de migration et à vous assurer que les tests sont exécutés sur une base de données cohérente avec l'état de votre base de données en direct.

Nous pouvons ajuster légèrement le gabarit create_test_database:

Commande d'importation alambiquée
de alembic.config import Config
Import App

...

@pytest.fixture(scope="session", autouse=True)
def create_test_database():
    url = str(app.DATABASE_URL)
    engine = create_engine(url)
    assert not database_exists(url),« Test database déjà existants.
    create_database(URL)             #Créer la base de test.
    config = Config("alembic.ini")   # Run the migrations.
    commande.upgrade(config, "head")
    Yield est un                            #Exécutez les tests.
    drop_database(url)               # Drop la base de test.