Lewati ke isi

Database

Starlette tidak sepenuhnya terkait dengan implementasi basis data tertentu.

Anda dapat menggunakannya dengan ORM asinkron seperti GINO, atau menggunakan titik akhir non-asinkron biasa dan berintegrasi dengan SQLAlchemy.

Dalam dokumen ini, kami akan menunjukkan cara menargetkan databases Integrasi paket, yang menyediakan dukungan inti SQLAlchemy untuk berbagai driver database yang berbeda.

Berikut adalah contoh lengkap yang mencakup definisi tabel, mengkonfigurasi instance database.Database, dan beberapa titik akhir yang berinteraksi dengan database.

**. Env **

DATABASE _ URL = sqlite: / / test.db

app.py

Kontekstual impor

Basis data impor
meter persegi import
Starlette.application diimpor dari Starlette
Konfigurasi impor dari starlette.config
Dari pemula. Tanggapan impor JSONResponse
Dari starlette. rute impor routing


# Konfigurasi dari variabel lingkungan atau file '.env'.
Konfigurasi = Konfigurasi ('.env')
DATABASE _ URL = konfigurasi ('DATABASE _ URL')


# Definisi tabel database.
Metadata = sqlalchemy. MetaData ()

Catatan = sqlalchemy. Tabel (
    "Catatan,"
    Metadata,
    Sqlalchemy. Kolom ("id," sqlalchemy. Integer, primary _ key = Benar),
    Sqlalchemy. Kolom ("teks," sqlalchemy. String),
    Sqlalchemy. Kolom ("selesai," sqlalchemy. Boolean),
)

Database = database. Database (DATABASE _ URL)

@ contextlib.asynccontextmanager
Async def umur (aplikasi):
    Tunggu koneksi database ()
    Hasil
    Tunggu database .putuskan sambungan ()

# Kode aplikasi utama.
Async def list _ notes (permintaan):
    Pertanyaan = notes.select ()
    Hasil = menunggu database. ambil _ semua (pertanyaan)
    Konten = [
        {
            "Teks": hasil ["teks"],
            "Selesai": hasil ["selesai"]
        }
        Untuk hasil dalam hasil
    ]
    Mengembalikan JSONResponse (konten)

Sinkronisasi def add _ note (permintaan):
    Data = menunggu request.json ()
    Query = notes.insert ().value (
       Teks = data ["teks"],
       Selesai = data ["selesai"]
    )
    Menunggu database .eksekusi (pertanyaan)
    Mengembalikan JSONResponse ({
        "Teks": data ["teks"],
        "Selesai": data ["selesai"]
    })

Rute = [
    Rute ("/ catatan," endpoint = list _ notes, metode = ["GET"]),
    Rute ("/ catatan," endpoint = add _ note, metode = ["POST"]),
]

Aplikasi = Starlette (
    Rute = rute,
    Umur = umur,
)

Terakhir, Anda perlu membuat database & tabel data. Alembic direkomendasikan dan kami menutupinya secara singkat dalam migrasi.

permintaan

Pertanyaan dapat dilakukan dengan menggunakan query SQLAlchemy Core.

Metode berikut didukung:

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

Transaksi; bisnis; transaksi (kata ini memiliki banyak arti, dan terjemahan spesifik tergantung pada konteksnya). Jika hanya kata "Transaksi" yang diberikan diterjemahkan secara harfiah, itu dapat diterjemahkan sebagai "transaksi; transaksi." Berikut adalah beberapa definisi umum untuk Anda pilih sesuai dengan konteks tertentu. Sesuai dengan persyaratan judul, output langsung diterjemahkan sebagai: transaksi

Transaksi database dapat digunakan sebagai dekorator, pengelola konteks, atau API tingkat rendah.

Gunakan dekorator pada titik akhir:

@ Database.transaksi ()
Async def mengisi _ note (permintaan):
    # Sisipan database ini terjadi dalam transaksi.
    # Ini akan digulung kembali oleh `RuntimeError`.
    Query = notes.insert ().value (text = "you won 't see me," selesai = Benar)
    Menunggu database .eksekusi (pertanyaan)
    Meningkatkan RuntimeError ()

Menggunakan manajer konteks:

Async def mengisi _ note (permintaan):
    Sinkronisasi dengan transaksi database ():
        # Sisipan database ini terjadi dalam transaksi.
        # Ini akan digulung kembali oleh `RuntimeError`.
        Query = notes.insert ().value (text = "you won 't see me," selesai = Benar)
        Menunggu permintaan.database .eksekusi (permintaan)
        Meningkatkan RuntimeError ()

Menggunakan API tingkat rendah:

Async def mengisi _ note (permintaan):
    Transaksi = menunggu Database.transaksi ()
    Coba:
        # Sisipan database ini terjadi dalam transaksi.
        # Ini akan digulung kembali oleh `RuntimeError`.
        Query = notes.insert ().value (text = "you won 't see me," selesai = Benar)
        Menunggu database .eksekusi (pertanyaan)
        Meningkatkan RuntimeError ()
    Kecuali:
        Menunggu transaksi.rollback ()
        Naikkan
    Lainnya:
        Menunggu transaksi.commit ()

uji isolasi

Saat menguji layanan yang menggunakan database, ada beberapa hal yang ingin kami pastikan. Persyaratan kami harus:

  • Uji dengan database terpisah.
  • Setiap kali tes di jalankan, database tes baru dibuat.
  • Pastikan status database antara setiap kasus tes terisolasi satu sama lain.

Untuk memenuhi persyaratan ini, kami perlu membangun aplikasi dan pengujian kami seperti ini:

Starlette.application diimpor dari Starlette
Konfigurasi impor dari starlette.config
Basis data impor

Konfigurasi = Konfigurasi (. "env")

TESTING = konfigurasi ('TESTING', cast = bool, default = Salah)
DATABASE _ URL = konfigurasi ('DATABASE _ URL', cast = database. Database seURL)
TEST _ DATABASE _ URL = DATABASE _ URL (database = 'test _' + DATABASE _ database URL)

# Gunakan 'force _ rollback' selama pengujian, untuk memastikan kami tidak melanjutkan perubahan database
# antara setiap kasus tes.
JIKA MENGUJI:
    Database = database. Database (TEST _ DATABASE _ URL, force _ rollback = Benar)
Lainnya:
    Database = database. Database (DATABASE _ URL)

Selama latihan berjalan, kita masih perlu mengatur TESTING dan mengatur database pengujian. Dengan asumsi kita menggunakan py.test, inilah tampilan conftest.py kita:

uji coba impor
Dari starlette.config import environ
Dari starlette.testclient impor TestClient
Dari sqlalchemy import create _ engine
Dari sqlalchemy _ memanfaatkan database impor _ ada, buat _ database, drop _ database

# Ini menetapkan `os.environ`, tetapi memberikan beberapa perlindungan tambahan.
# Jika kami menempatkannya di bawah impor aplikasi, itu akan meningkatkan kesalahan
# memberi tahu kami bahwa 'PENGUJIAN' telah dibaca dari lingkungan.
Environ ['PENGUJIAN'] = 'Benar'

Aplikasi impor


@ Pytest.fixture (scope = "sesi," autouse = Benar)
Def membuat _ test _ database ():
  """
  Buat database yang bersih pada setiap kasus pengujian.
  Demi keamanan, kita harus batalkan jika sudah ada database.

  Kami menggunakan paket `sqlalchemy_utils` di sini untuk beberapa pembantu secara konsisten
  Membuat dan menjatuhkan database.
  """
  Url = str (aplikasi. UJI _ DATABASE _ URL)
  Mesin = buat _ mesin (url)
  Tegaskan tidak database _ ada (url), 'Uji database sudah ada. Aborting tes.'
  buat _ database (url) # Buat database pengujian.
  Metadata. buat _ semua (mesin) # Buat tabel.
  Hasil # Jalankan tes.
  drop _ database (url) # Jatuhkan database tes.


@ pytest.fixture ()
Pasti klien ():
    """
    Saat menggunakan perlengkapan 'klien' dalam kasus uji, kami akan mendapatkan database lengkap
    Pengembalian antar kasus uji:

    Tes pasti _ beranda (klien):
        Url = app.url _ path _ untuk ('beranda')
        Tanggapan = klien. Dapatkan (url)
        Tegaskan tanggapan. status _ code = = 200
    """
    Dengan TestClient (aplikasi) sebagai klien:
        Menghasilkan klien

Migrasi

Untuk mengelola perubahan tambahan pada database Anda, Anda hampir pasti perlu menggunakan migrasi database. Untuk ini, kami sangat merekomendasikan Alembic, yang ditulis oleh penulis SQLAlchemy.

$Pip instal alembic
$alembic init migrasi

Sekarang Anda perlu mengaturnya sehingga Alembic mereferensikan URL DATABASE _ yang dikonfigurasi dan menggunakan metadata tabel Anda.

Hapus baris berikut di alembic.ini:

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

Dalam migrations/env.py, Anda perlu mengatur kunci konfigurasi 'sqlalchemy.url' dan variabel target_metadata. Anda akan membutuhkan sesuatu seperti ini:

# Objek Konfigurasi Alembik.
Konteks = kontekstut.config

# Konfigurasikan Alembic untuk menggunakan URL DATABASE _ kami dan definisi tabel kami...
Aplikasi impor
Config.set _ main _ option ('sqlalchemy.url', str (app.DATABASE _ URL))
target _ metadata = app.metadata

...

Kemudian, dengan menggunakan contoh catatan kami di atas, buat revisi awal:

Revisi alembik -m "Buat tabel catatan"

Dan isi file baru (di dalam migrations/versions) dengan instruksi yang diperlukan:

Upgrade pasti ():
    Pilih. membuat _ tabel (
      'Catatan',
      Sqlalchemy. Kolom ("id," sqlalchemy. Integer, primary _ key = Benar),
      Sqlalchemy. Kolom ("teks," sqlalchemy. String),
      Sqlalchemy. Kolom ("selesai," sqlalchemy. Boolean),
    )

Benar-benar downgrade ():
    Op. drop _ table ('catatan')

Dan jalankan migrasi pertama Anda. Aplikasi pencatat kami sekarang aktif dan berjalan!

Kepala peningkatan alembik

Jalankan migrasi selama pengujian

Ini adalah praktik yang baik untuk memastikan bahwa rangkaian pengujian Anda menjalankan migrasi basis data setiap kali Anda membuat basis data pengujian. Ini akan membantu mengidentifikasi masalah apa pun dalam skrip migrasi Anda dan membantu memastikan bahwa tes di jalankan terhadap basis data yang konsisten dengan status basis data langsung Anda.

Kami dapat menyesuaikan perlengkapan create_test_database sedikit:

Dari perintah impor alembic
Konfigurasi impor dari alembic.config
Aplikasi impor

...

@ Pytest.fixture (scope = "sesi," autouse = Benar)
Def membuat _ test _ database ():
    Url = str (aplikasi. DATABASE _ URL)
    Mesin = buat _ mesin (url)
    Tegaskan tidak database _ ada (url), 'Uji database sudah ada. Aborting tes.'
    buat _ database (url) # Buat database pengujian.
    Konfigurasi = Konfigurasi ("alembic.ini") # Jalankan migrasi.
    Perintah.upgrade (konfigurasi, "kepala")
    Hasil # Jalankan tes.
    drop _ database (url) # Jatuhkan database tes.