Zum Inhalt

Datenbank

Starlette ist nicht ausschließlich an eine bestimmte Implementierung der Datenbank gebunden.

Sie können es mit asynchronen ORMs wie GINO oder mit normalen, nicht asynchronen Endpunkten verwenden, die in SQLAlchemy integriert sind.

In diesem Dokument zeigen wir Ihnen, wie Sie ein databases-Paket integrieren, das den SQLAlchemy-Kernel für eine Reihe von verschiedenen Datenbanktreibern unterstützt.

Hier ist ein vollständiges Beispiel mit einer Tabellendefinition, der Konfiguration einer Instanz von database.Database und mehreren Endpunkten, die mit der Datenbank interagieren.

**und Env **

DATABASE_URL = sqlite:///test.db

app.py

Import von Kontext

Import von Datenbanken
Import von Sqlalchemy
von starlette. applications import Starlette
from starlette. config import Config (Eingabetaste)
von starlette. responses import JSONResponse
von starlette. routing import Route


#Konfiguration von Umgebungsvariablen oder Datei '. env'
config = Config ('. env')(Übersetzung)
DATABASE_URL = config ('DATASE_URL')


Definition einer Tabelle.
metadata = sqlalchemy. metadata ()[Bearbeiten | Quelltext bearbeiten]

Die Tabelle mit der Tabelle "(
    Die „ Noten ",
    metadaten,
    sqlalchemy. Column ("id", sqlalchemy. Integer, primary_key = true)("id", sqlalchemy. Integer, primary_key = true), und
    sqlalchemy. Column ("text", sqlalchemy. String),
    sqlalchemy. Column ("completed", sqlalchemy. Boolean), und
)

Datenbank = Datenbank . Datenbank (DATABASE _ URL)

@contextlib . asynccontextmanager für
async def lifespan (Englisch) :
    Siehe auch : await database . connect ()
    und yield
    await database . disconnect () (Deutsche Übersetzung)

# Hauptanwendungscode .
async def list _ notes (Anforderung) :
    Die Query =notes.select()
    results = await database . fetch _ all (query) [ Bearbeiten | Quelltext bearbeiten ]
    Inhalt [ ]
        {
            " Text " : Ergebnis [ " Text " ] ,
            " abgeschlossen " : Ergebnis [ " abgeschlossen " ]
        }
        für Ergebnisse in Results
    ]
    Zurück zu JSONResponse (Content)

async def add _ note (Anforderung) :
    Frage von json = await request ()
    Die Funktion query () . values (
       text = Daten [ " Text " ] ,
       Komplettiert = Daten [ " completed " ]
    )
    Die Datei Datei wird aufgerufen . execute (query)
    return JSONResponse ({Vorlage : JSONResponse (
        " Text " : Daten [ " Text " ] ,
        " completed " (abgeschlossen) : data [ " completed " ]
    })

Die Bahn [ … ]
    Route ("/notes", endpoint = list_notes, methods =["GET"]), und
    Route ("/notices", endpoint = add_note, methods =["POST"]),
]

Die Startelf (
    Straßen = Routen,
    Lebenszeichen = Lifespan,
)

Schließlich sollten Sie eine Datenbanktabelle erstellen, die wir im Rahmen der Migration kurz beschrieben haben.

Befragung

Abfragen können mit SQLAlchemy Core-Abfragen durchgeführt werden.

Folgende Methoden werden unterstützt:

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

Transaktion; Geschäft; Transaktion (das Wort hat mehrere Bedeutungen und die spezifische Übersetzung muss je nach Kontext bestimmt werden), wenn es nur als "Transaktion; Transaktion" übersetzt wird, werden hier einige gemeinsame Umschreibungen für Sie bereitgestellt, damit Sie die Übersetzung für die direkte Ausgabe nach dem jeweiligen Kontext auswählen können, wie vom Thema gefordert: Transaktion.

Datenbanktransaktionen können als Dekoratoren, Kontextmanager oder Low-Level-APIs verwendet werden.

Verwendung von Dekoratoren am Ende:

Beispiel: database. transaction ()
async def populate_note (Anforderung):
    #Diese Datenbank enthält Verbindungen innerhalb einer Transaktion.
    Es wird durch die zurückrollende `RuntimeError`.
    query = notices. insert (). values (text ="Sie werden mich nicht sehen", completed = True)
    Die Datei Datei wird aufgerufen. execute (query)
    raise runtimeError ()(Übersetzung)

Verwendung des Download-Managers:

async def populate_note (Anforderung):
    async mit der Datenbank. transaction ():
        #Diese Datenbank enthält Verbindungen innerhalb einer Transaktion.
        #Es wird von `RuntimeError` zurückgerollt.
        query = notices. insert (). values (text ="Sie werden mich nicht sehen", completed = True)
        Abfrage, Datenbank, Abfrage (Query)
        raise runtimeError ()(Übersetzung)

Verwendung einer Low-Level-API:

async def populate_note (Anforderung):
    transaction = await database. transaction ()(= die Datei wird angezeigt)
    versuchen:
        #Diese Datenbank enthält Verbindungen innerhalb einer Transaktion.
        #Es wird durch `RuntimeError` zurückgerollt.
        query = notices. insert (). values (text ="Sie werden mich nicht sehen", completed = True)
        Die Datei Datei wird aufgerufen. execute (query)
        raise runtimeError ()(Übersetzung)
    ausgenommen:
        await transaction. rollback ()(Übersetzung)
        Raise aufziehen
    und Else:
        Aktion. commit ()

Test isoliert

Bei der Prüfung von Diensten, die Datenbanken verwenden, gibt es ein paar Dinge, die wir sicherstellen möchten, dass unsere Anforderungen lauten:

  • Verwenden Sie eine separate Datenbank für die Tests.
  • Bei jeder Testdurchführung wird eine neue Testdatenbank erstellt.
  • Stellen Sie sicher, dass die Datenbankzustände zwischen den einzelnen Testfällen isoliert sind.

Um diesen Anforderungen gerecht zu werden, müssen wir unsere Anwendungen und Tests wie folgt entwickeln:

von starlette. applications import Starlette
from starlette. config import Config (Eingabetaste)
Import von Datenbanken

config = Config (". env") Bearbeiten

TESTING = config ('TESTING', cast = bool, default = False) und
DATABASE_URL = config ('DATABASE_URL', cast = databases. DatabaseURL) oder
TEST_DATABASE_URL = DATABASE_URL. replace (database ='test_'+ DATABASE_URL. database)[Bearbeiten | Quelltext bearbeiten]

#Verwenden Sie 'force_rollback' während des Tests, um sicherzustellen, dass wir keine Änderungen an der persönlichen Datenbank vornehmen
zwischen jedem Testfall.
if testing schreibt:
    Datenbank = Datenbank. Datenbank (TEST_DATABASE_URL, force_rollback = True)
und Else:
    database = Datenbank. Datenbank (DATABASE_URL)

Während des Testlaufs müssen wir noch TESTING einrichten und die Testdatenbank einrichten, vorausgesetzt, wir verwenden py.test, und hier ist, wie unser conftest.py aussehen könnte:

Der Import von Pytest
from starlette. config import environ (englisch)
von starlette. testclient import TestClient
from sqlchemy import create_engine ()[Bearbeiten]
from sqlalchemy_utils import database_exists, create_database und drop_database

#Dies setzt `os.environ` ein, bietet aber zusätzlichen Schutz.
#Wenn wir es unter den Import der Anwendung gestellt haben, hätte es einen Fehler verursacht
#informing us that 'testing' had already been read from the environment.#informativ, dass 'testing' bereits aus der Umwelt gelesen wurde.
[[Kategorie: Hochhaus | Hochhaus]]="True"

Die Import-App


@ pytest. fixture (scope ="session", autouse = true)@ pytest. fixture (scope ="Sitzung", autouse = true)
def create_test_database ()(zum Beispiel):
  """
  Erstellen Sie eine saubere Datenbank in jedem Testfall.
  Aus Sicherheitsgründen sollten wir abort, wenn eine Datenbank bereits vorhanden ist.

  Wir verwenden das `sqlalchemy_utils`-Paket hier für ein paar Helfer konsistently
  Erstellen und Droppen der Datenbank.
  """
  url = str (app. TEST_DATABASE_URL)[Bearbeiten | Quelltext bearbeiten]
  engine = create_engine (url) Bearbeiten
  assert not database_exists (url),'Testdatenbank bereits vorhanden.
  create_database (url) erzeugen             #Erstellen Sie die Testdatenbank.
  metadata. create_all (Englisch)      Erstellen Sie die Tabellen
  und yield                            #Führen Sie die Tests aus.
  Die Datei drop_database (url)               Klicken Sie auf die Testdatenbank.


@ pytest. fixture ()(auf Englisch)
def der Client ():
    """
    Wenn wir die "Client"-Fixture in Testfällen verwenden, erhalten wir die vollständige Datenbank.
    Rollbacks zwischen den Test-Cases:

    def test_homepage (client)(auf Englisch):
        url = app. url_path_for ('homepage') url = app. url_path_for ('Homepage ')
        Antwort = client. get (URL)
        assert response. status_code = 200 Zeichen
    """
    Mit TestClient (app) als Client:
        Der yield client

Auswandern; Auswandern

Um inkrementelle Änderungen an einer Datenbank zu verwalten, müssen Sie mit ziemlicher Sicherheit eine Datenbankmigration verwenden Um dies zu tun, empfehlen wir dringend Alembic, geschrieben von den Autoren von SQLAlchemy.

$pip install alembic für Mac
$alembic init migrations {pl}{pl}

Jetzt müssen Sie dies so einrichten, dass Alembic die konfigurierte DATABASE_URL referenziert und Ihre Tabellenmetadaten verwendet.

Löschen Sie die folgende Zeile in alembic.ini:

sqlalchemy. url = Fahrer://Benutzer: pass @ localhost/dbname

In migrations/env.py müssen Sie den 'sqlalchemy.url'-Konfigurationstasten und die target_metadata-Variable so setzen:

Das alembic config object.
config = context. config = Kontextdaten

Configure Alembic to use our DATABASE_URL and our table definitions (Konfigurieren Sie Alembic, um unsere DATABASE_URL und unsere Tabellendefinitionen zu verwenden)
Die Import-App
config. set_main_option ('sqlalchemy. url', str (app. DATABASE_URL))[Bearbeiten | Quelltext bearbeiten]
target_metadata = App. metadata (engl.)

...

Verwenden Sie dann unser Beispiel für die oben genannten Notizen, um eine erste überarbeitete Version zu erstellen:

alembic revision-m "Erstellen einer Tabelle"

Füllen Sie die neue Datei (in migrations/versions) mit den notwendigen Anweisungen:

def upgrade (): zum Vergrößern
    Der Tisch (
      Die „ Noten ",
      sqlalchemy. Column ("id", sqlalchemy. Integer, primary_key = true)("id", sqlalchemy. Integer, primary_key = true), und
      sqlalchemy. Column ("text", sqlalchemy. String),
      sqlalchemy. Column ("completed", sqlalchemy. Boolean), und
    )

def downgrade (): def downgrade ()(zum Vergrößern)
    p. drop_table ('Noten ')

Und führen Sie Ihre erste Migration durch Unsere Notizen-App läuft jetzt!

Erläuterung Alembic Upgrade Head

Umzug während der Testphase durchführen

Es ist eine gute Praxis, jedes Mal, wenn Sie eine Testdatenbank erstellen, sicherzustellen, dass Ihre Testsuite eine Datenbankmigration ausführt Dies wird dazu beitragen, Probleme in Ihrem Migrationsskript zu erkennen und sicherzustellen, dass die Tests für die Datenbank ausgeführt werden, die mit Ihrem Echtzeit-Datenbankstatus übereinstimmt.

Wir können die create_test_database-Klemme etwas anpassen:

von alembic import command
from alembic. config import Config (Eingabeaufforderung)
Die Import App

...

@ pytest. fixture (scope ="session", autouse = true)@ pytest. fixture (scope ="Sitzung", autouse = true)
def create_test_database ()(zum Beispiel):
    url = str (Anwendung. DATABASE_URL)
    engine = create_engine (url) Bearbeiten
    assert not database_exists (url),'Testdatenbank bereits vorhanden.
    create_database (url) erzeugen             #Erstellen Sie die Testdatenbank.
    config = Config ("alembic.ini") Bearbeiten   Laufen Sie die Migrationen
    Befehl. upgrade (config,"head") Bearbeiten
    und yield                            #Führen Sie die Tests aus.
    Die Datei drop_database (url)               Klicken Sie auf die Testdatenbank.