Перейти к содержанию

База данных

Starlette не привязана к какой-либо конкретной реализации базы данных.

Вы можете использовать его с асинхронным ORM (например, GINO) или использовать обычные неасинхронные конечные точки, интегрированные с SQLALchemy.

В этом документе мы покажем, как интегрировать пакет databases, который обеспечивает основную поддержку SQLALchemy для широкого спектра драйверов баз данных.

Вот полный пример, который включает определение таблицы, конфигурацию экземпляра database.Database и несколько конечных точек, взаимодействующих с базой данных.

**. env **

DATABASE_URL = sqlite://test.db

app.py

Импорт contextlib

Import Data abases
Sqimport alchemy
from Starlette. applications import Starlette
Из starlet.config import Config
from starlette. response import JSONResponse
От Starlet.routing Import Route


#Configuration from environment variables или '. env' file.
config = Config ('. env ')
DATABASATA_URL = config ('DATABASE_URL ')


#Database table definitions
Metadata = sqalchemy. MetaData

(((
    «Notes»,
    метаданные,
    sqlalchemy. Column (id, sqlalchemy. Integer, primary_key = True)
    sqlalchemy («текст», sqlalchemy. String),
    sqlalchemy («completed», sqlalchemy, Boolean)
)

Database = databases. Databases (DATABASE)

Contextlib.asynccontextmanager
async def lifespan(app):
    database.connect(
    Yield
    database.disconnect()

Основной код приложения.
async def list_notes (request)
    Query = notes.select (
    results = await database. fetch_all (query)
    content = [
        {
            «текст»: результат [«текст»],
            "completed": result ("completed")
        }
        Результаты в results
    ]
    Return JSONResponse(контент)

async def add_note (request):
    data = await request.json ()
    (((
       text=data["text"],
       completed=data["completed"]
    )
    await database.execute(query)
    Ремонт ((
        "text": data["text"],
        "completed": data["completed"]
    })

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

Слайд (
    routes=routes,
    lifespan=lifespan,
)

Наконец, вам нужно создать таблицу базы данных. Рекомендуется использовать Alembic, который мы кратко описали в миграции.

Запрос

Запросы можно выполнять с помощью SQLALchemy Core.

Поддерживаются следующие методы:

  • 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 переводится как «Трансакции», оно переводится как «Транспорт». Здесь вы найдете несколько общих толкований, которые вы можете выбрать в зависимости от контекста. В зависимости от темы, прямой вывод будет следующим: Торговля

Транзакции базы данных могут использоваться в качестве декоратора, контекстного менеджера или низкоуровневого API.

Используйте отделку на конечной точке:

database.transaction()
async def populate_note (request):
    # This database insert occurs within a transaction
    It will be rolled back by the `RuntimeError`.
    Query = notes.insert ().values(text="You Won't See Me", completed=True)
    await database.execute(query)
    RuntimeError(

Использование контекстного диспетчера:

async def populate_note (request):
    async with database.transaction():
        # This database insert occurs within a transaction
        It will be rolled back by the `RuntimeError`.
        Query = notes.insert ().values(text="You Won't See Me", completed=True)
        database execute(query)
        RuntimeError(

Использование API низкого уровня:

async def populate_note (request):
    transaction = await database.transaction(
    try:
        # This database insert occurs within a transaction
        It will be rolled back by the `RuntimeError`.
        Query = notes.insert ().values(text="You Won't See Me", completed=True)
        await database.execute(query)
        RuntimeError(
    except:
        Пересечение (()
        Rise
    else:
        Переводчик ()

Тестирование изоляции

При тестировании сервисов, использующих базу данных, мы хотим убедиться в том, что у нас есть несколько вещей. Наши требования:

  • Тестирование проводится с использованием отдельной базы данных.
  • При каждом запуске тестирования создается новая база данных тестирования.
  • Убедитесь, что состояние базы данных в каждом тестовом случае изолировано друг от друга.

Чтобы соответствовать этим требованиям, нам нужно построить наши приложения и тесты следующим образом:

from Starlette.applications import Starlette
Из starlet.config import Config
Import Data abases

config = Config(".env")

TESTING = config ('TESTING', cast=bool, default=False)
DATABASE_URL = config ('DATA_URL', cast= databases.DatabaseURL)
TEST_DATABASE_URL = DATABASE_URL.replace (database ='test_' + DATABASE URL.database)

# Use 'force_rollback' during testing, to ensure we do not persist database changes
#Между тем тест по делам.
If Testing:
    Database = databases.Database(TEST_DATASATA_URL, force_rollback =True)
else:
    Database = databases. Databases (DATABASE)

Во время пробного запуска нам все еще нужно настроить TESTING и настроить тестовую базу данных. Предположим, что мы используем py.test, вот, как будет выглядеть наш conftest.py:

Импорт Pytest
Из starlette.config import environ
Testclient Import TestClient
Из sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database

# This sets `os.environ`, but provides is not addtional protection.
If We placed it below the application import, it would raise an error
# informing us that 'TESTING' had already been read from the environment.
environ ('Testing'] = 'True'

Приложение import App


pytest.fixture (scope="session", autouse=True)
create def_test_database ():
  """
  Создать чистую базу данных на every test case.
  Для safety, мы должны сделать это, если база данных already exists.

  We use the `sqlalchemy_utils` package here for a few helpers in consistently
  Creating and dropping the database.
  """
  url = str(ap.tEST_DATASATA_URL)
  Engine = create_engine (url)
  assert not database_exists (url), 'Test database already exists. Aborting tests.'
  create_database (url)             Создать базу данных тестирования.
  metadata. create_all (engine)      Создать табл.
  Yield                            Run the Tests.
  drop_database (url)               # Drop the test database


Pytest.fixture ()
Def client():
    """
    When using the 'client' fixture in test cases, we'll get full database
    rollbacks between test cases:

    def test_homepage (клиент):
        url = app.url_path_for ('homepage')
        Ответ = client.get (url)
        assert response. status_code == 200
    """
    with TestClient(приложение) as client:
        Клиент Yield

Миграция; миграция

Для управления инкрементными изменениями в базе данных вам почти наверняка придется использовать миграцию базы данных. Для этого мы настоятельно рекомендуем Alembic, который был написан автором SQLALchemy.

$ pip install alembic
$ alembic init migrations

Теперь вам нужно настроить, чтобы Alembic ссылался на настроенный DATABASE URL-адрес и использовал метаданные таблицы.

Удалите следующую строку в alembic.ini:

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

В migrations/env.py вам нужно настроить ключ конфигурации 'sqlalchemy.url' и переменную target_metadata. Вам понадобится что-то вроде этого:

The Alembic Config Object.
config = config

# Configure Alembic to use our DATABASE_URL and our table definitions
Приложение import App
config. set_main_option ('sqlalchemy.url', str(app. DATABASATA_URL))
target_metadata = app.metadata

...

Затем создайте предварительную редакцию, используя наш пример примечаний выше:

Alembic revision -«Create Notes table»

Заполните новый файл необходимыми инструкциями (внутри migrations/versions):

def upgrade():
    Колонна (
      «Notes»,
      sqlalchemy.Column(id, sqlalchemy.Integer, primary_key =True)
      sqlalchemy («текст», sqlalchemy.String),
      sqlalchemy («completed», sqlalchemy, Boolean)
    )

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

И запустите свой первый переход. Наше приложение для заметок теперь работает!

Alembic Upgrade head

Запуск миграции во время тестирования

При создании тестовой базы данных рекомендуется убедиться, что ваш тестовый пакет запускает миграцию базы данных. Это поможет выявить любые проблемы в вашем сценарии миграции и поможет убедиться, что тестирование выполняется для базы данных, которая соответствует состоянию базы данных в реальном времени.

Мы можем немного настроить create_test_database:

Из Alembic Import Command
Из alembic.config import Config
Приложение Import App

...

pytest.fixture (scope="session", autouse=True)
create def_test_database ():
    url = str(app. DATABASATA_URL)
    Engine = create_engine (url)
    assert not database_exists (url), 'Test database already exists. Aborting tests.'
    create_database (url)             Создать базу данных тестирования.
    config = Config("alembic.ini")   Run the migrations.
    command.upgrade(config, "head")
    Yield                            Run the Tests.
    drop_database (url)               # Drop the test database