Ga naar inhoud

Database

Starlette is niet strikt gebonden aan een bepaalde database-implementatie.

Je kunt het gebruiken met asynchrone ORM 's zoals GINO, of gewone niet-asynchrone eindpunten gebruiken en integreren met SQLAlchemy.

In dit document laten we zien hoe u kunt integreren met het databases-pakket, dat SQLAlchemy-kernondersteuning biedt voor een reeks verschillende databasestuurprogramma 's.

Hier is een compleet voorbeeld met tabeldefinitie, het configureren van een database.Database-instantie en verschillende eindpunten die interactie hebben met de database.

**. Env **

DATABASE _ URL = sqlite: / / test.db

app.py

Contextlib importeren

Databanken importeren
Sqlalchemie invoeren
Starlette.applicaties importeren vanuit Starlette
Config importeren vanuit starlette.config
Uit starlette.reacties importeren JSONResponse
Van starlette. routing importeren Route


# Configuratie van omgevingsvariabelen of '.env' -bestanden.
Config = Config ('.env')
DATABASE _ URL = configuratie ('DATABASE _ URL')


# Database tabel definities.
Metagegevens = sqlalchemie. MetaData ()

Opmerkingen = sqlalchemie. Tabel (
    "Aantekeningen",
    Metadata,
    Sqlalchemie. Kolom ("id", sqlalchemie. Geheel getal, primaire _ sleutel = Waar),
    Sqlalchemie. Kolom ("tekst", sqlalchemie. String),
    Sqlalchemie. Kolom ("voltooid", sqlalchemie. Booleaans),
)

Database = databanken. Database (DATABASE _ URL)

@ contextlib.asynccontextmanager
Async def levensduur (app):
    Wacht op database connect ()
    Opbrengst
    Database afwachten .disconnect ()

# Hoofdtoepassingscode.
Async def list _ notes (verzoek):
    Query = notes.select ()
    Resultaten = wacht database. fetch _ all (query)
    Inhoud = [
        {
            "Tekst": resultaat ["tekst"],
            "Voltooid": resultaat ["voltooid"]
        }
        Voor resultaten in resultaten
    ]
    Retourneert JSONResponse (inhoud)

Async def add _ note (verzoek):
    Data = wacht op aanvraag.json ()
    Query = notes.insert () .waarden (
       Tekst = gegevens ["tekst"],
       Voltooid = gegevens ["voltooid"]
    )
    In afwachting van database. Uitvoeren (query)
    Geeft JSONResponse ({
        "tekst": gegevens ["tekst"],
        "Voltooid": gegevens ["voltooid"]
    })

Routes = [
    Route ("/ notes", eindpunt = list _ notes, methoden = ["GET"]),
    Route ("/ notes", eindpunt = add _ note, methoden = ["POST"]),
]

Uitgever: App = Starlette (
    Routes = routes,
    Levensduur = levensduur,
)

Ten slotte moet u de gegevensdatabase en tabel maken. Alembic wordt aanbevolen en we behandelen het kort in de migratie.

zoekopdracht

Query 's kunnen worden uitgevoerd met SQLAlchemy Core-query 's.

De volgende methoden worden ondersteund:

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

Transactie; bedrijf; transactie (dit woord heeft meerdere betekenissen en de specifieke vertaling is afhankelijk van de context). Als alleen het gegeven woord "Transacties" letterlijk wordt vertaald, kan het worden vertaald als "transactie; transactie". Hier zijn verschillende algemene definities die u kunt kiezen op basis van de specifieke context. Volgens de vereisten van de titel wordt de directe uitvoer vertaald als: transactie

Databasetransacties kunnen worden gebruikt als decorateurs, contextmanagers of low-level API 's.

Gebruik decorateurs op eindpunten:

@ Database.transactie ()
Async def populate _ note (verzoek):
    # Deze database-invoeging vindt plaats binnen een transactie.
    # Het wordt teruggedraaid door de `RuntimeError`.
    Query = notes.insert () .waarden (text = "je zult me niet zien", voltooid = Waar)
    In afwachting van database. Uitvoeren (query)
    RuntimeError verhogen ()

De contextmanager gebruiken:

Async def populate _ note (verzoek):
    Async met database transactie ():
        # Deze database-invoeging vindt plaats binnen een transactie.
        # Het wordt teruggedraaid door de `RuntimeError`.
        Query = notes.insert () .waarden (text = "je zult me niet zien", voltooid = Waar)
        In afwachting van het verzoek. Databases. Uitvoeren (query)
        RuntimeError verhogen ()

Een low-level API gebruiken:

Async def populate _ note (verzoek):
    Transactie = wachten Database.transactie ()
    Probeer:
        # Deze database-invoeging vindt plaats binnen een transactie.
        # Het wordt teruggedraaid door de `RuntimeError`.
        Query = notes.insert () .waarden (text = "je zult me niet zien", voltooid = Waar)
        In afwachting van database. Uitvoeren (query)
        RuntimeError verhogen ()
    Behalve:
        In afwachting van transaction.rollback ()
        Verhogen
    Anders:
        In afwachting van transaction.commit ()

test isolatie

Bij het testen tegen een service die een database gebruikt, zijn er een paar dingen die we willen garanderen. Onze vereisten moeten zijn:

  • Test met een aparte database.
  • Elke keer dat een test wordt uitgevoerd, wordt er een nieuwe testdatabase gemaakt.
  • Zorg ervoor dat de databasestatus tussen elke testcase van elkaar is geïsoleerd.

Om aan deze vereisten te voldoen, moeten we onze applicaties en tests als volgt bouwen:

Starlette.applicaties importeren vanuit Starlette
Config importeren vanuit starlette.config
Databanken importeren

Config = Config (".env")

TESTEN = config ('TESTEN', cast = bool, standaard = Onwaar)
DATABASE _ URL = config ('DATABASE _ URL', cast = databases. DatabaseURL)
TEST _ DATABASE _ URL = DATABASE _ URL (database = 'test _' + DATABASE _ URL database)

# Gebruik 'force _ rollback' tijdens het testen om ervoor te zorgen dat we databasewijzigingen niet voortzetten
# tussen elk testgeval.
BIJ HET TESTEN:
    Database = databanken. Database (TEST _ DATABASE _ URL, force _ rollback = Waar)
Anders:
    Database = databanken. Database (DATABASE _ URL)

Tijdens de oefenrun moeten we nog TESTING opzetten en de testdatabase opzetten. Ervan uitgaande dat we py.test gebruiken, is dit hoe onze conftest.py eruit zou kunnen zien:

Pytest importeren
Van starlette.config importeren environ
Vanaf starlette.testclient importeren TestClient
Vanuit sqlalchemy importeren create _ engine
Vanuit sqlalchemy _ utils import database _ bestaat, create _ database, drop _ database

# Dit stelt `os.environ` in, maar biedt wat extra bescherming.
# Als we het onder de applicatie-import zouden plaatsen, zou er een fout optreden
# informeerde ons dat 'TESTING' al uit de omgeving was gelezen.
Environ ['TESTEN'] = 'Waar'

App importeren


@ Pytest.fixture (scope = "sessie", automatisch = Waar)
Maak _ test _ database aan ():
  """
  Maak een schone database voor elke testcase.
  Voor de veiligheid moeten we afbreken als er al een database bestaat.

  We gebruiken hier het `sqlalchemy_utils` pakket voor een paar helpers in consequent
  Het creëren en verwijderen van de database.
  """
  Url = str (app. TEST _ DATABASE _ URL)
  Motor = aanmaken _ engine (url)
  Beweren niet database _ bestaat (url), 'Testdatabase bestaat al. Afbreken van tests.'
  create _ database (url) # De testdatabase maken.
  Metadata. create _ all (engine) # De tabellen maken.
  Opbrengst # Voer de tests uit.
  drop _ database (url) # Laat de testdatabase vallen.


@ pytest.armatuur ()
Def cliënt ():
    """
    Wanneer we de 'client' -armatuur gebruiken in testgevallen, krijgen we de volledige database
    Rollbacks tussen testgevallen:

    Def test _ startpagina (client):
        Url = app.url _ path _ voor ('startpagina')
        Reactie = client.get (url)
        Antwoord. status _ code = = 200
    """
    Met TestClient (app) als client:
        Opbrengst opdrachtgever

Migratie

Om incrementele wijzigingen in uw database te beheren, moet u vrijwel zeker databasemigratie gebruiken. Hiervoor raden we Alembic ten zeerste aan, geschreven door de auteurs van SQLAlchemy.

$Pip installeert alembic
$alembic init migraties

Nu moet u het zo instellen dat Alembic verwijst naar de geconfigureerde DATABASE _ URL en uw tabelmetadata gebruikt.

Verwijder de volgende regels in alembic.ini:

Sqlalchemy.url = stuurprogramma: / / gebruiker: pass@localhost/ dbname

In migrations/env.py moet je de 'sqlalchemy.url' configuratiesleutel en de target_metadata variabele instellen. Je hebt zoiets als dit nodig:

# Het Alembic Config-object.
Context = context.config

# Configureer Alembic om onze DATABASE _ URL en onze tabeldefinities te gebruiken...
App importeren
Config.set _ main _ option ('sqlalchemy.url', str (app.DATABASE _ URL))
doel _ metadata = app.metadata

...

Maak vervolgens met behulp van ons notitievoorbeeld hierboven een voorlopige herziening:

Alembic revisie -m "Notitietabel maken"

En vul het nieuwe bestand (binnen migrations/versions) met de nodige instructies:

Def upgrade ():
    Op. _ tabel aanmaken (
      'Aantekeningen',
      Sqlalchemie. Kolom ("id", sqlalchemie. Geheel getal, primaire _ sleutel = Waar),
      Sqlalchemie. Kolom ("tekst", sqlalchemie. String),
      Sqlalchemie. Kolom ("voltooid", sqlalchemie. Booleaans),
    )

Zeker downgraden ():
    Op. drop _ table ('noten')

En voer uw eerste migratie uit. Onze notitie-app is nu actief!

Alembic upgrade hoofd

Voer de migratie uit tijdens het testen

Het is een goede gewoonte om ervoor te zorgen dat uw testsuite elke keer dat u een testdatabase maakt een databasemigratie uitvoert. Dit helpt bij het identificeren van problemen in uw migratiescripts en helpt ervoor te zorgen dat tests worden uitgevoerd tegen een database die consistent is met de status van uw live database.

We kunnen het create_test_database armatuur een beetje aanpassen:

Van alembic import opdracht
Config importeren vanuit alembic.config
App importeren

...

@ Pytest.fixture (scope = "sessie", automatisch = Waar)
Maak _ test _ database aan ():
    Url = str (app. DATABASE _ URL)
    Motor = aanmaken _ engine (url)
    Beweren niet database _ bestaat (url), 'Testdatabase bestaat al. Afbreken van tests.'
    create _ database (url) # De testdatabase maken.
    Config = Config ("alembic.ini") # Voer de migraties uit.
    Command.upgrade (config, "hoofd")
    Opbrengst # Voer de tests uit.
    drop _ database (url) # Laat de testdatabase vallen.