콘텐츠로 이동

데이터베이스

Starlette는 특정 데이터베이스 구현과 엄격하게 연결되어 있지 않습니다.

GINO와 같은 비동기 ORM과 함께 사용하거나 일반 비동기 엔드포인트를 사용하여 SQLAlchemy와 통합할 수 있습니다.

이 문서에서는 다양한 데이터베이스 드라이버에 대해 SQLAlchemy 핵심 지원을 제공하는 databases 패키지에 대해 통합하는 방법을 보여 줍니다.

다음은 테이블 정의,database.Database 인스턴스 구성 및 데이터베이스와 상호 작용하는 여러 엔드포인트를 포함하는 전체 예입니다.

**. Env **

DATABASE_URL = sqlite://test.db

app.py

컨텍스트 립 가져오기

데이터베이스 가져오기
sqlalchemy 가져오기
Starlette. 응용 프로그램이 Starlette에서 가져옵니다.
Starlette.config에서 가져오기 구성
Starlette.에서 JSONResponse 가져오기
별자리에서. 라우팅 가져오기 경로


# 환경 변수 또는 '.env' 파일의 구성.
구성 = 구성('.env')
DATABASE_URL = 구성('DATABASE_URL')


# 데이터베이스 테이블 정의.
메타데이터 = sqlalchemy. 메타데이터()

참고 = sqlalchemy. 테이블(
    "노트",
    메타데이터,
    Sqlalchemy. 열("id", sqlalchemy. 정수, prial_key = 참),
    Sqlalchemy. 열("텍스트", sqlalchemy. 문자열),
    Sqlalchemy. 열("완료", sqlalchemy. 부울),
)

데이터베이스 = 데이터베이스. 데이터베이스(DATABASE_URL)

@
비동기 def 수명(앱):
    데이터베이스 연결 대기()
    수율
    데이터베이스를 기다립니다.연결 끊기()

# 기본 응용 프로그램 코드.
비동기 def list_(요청):
    쿼리=notes.select()
    결과 = 데이터베이스를 기다립니다.
    내용 = [
        {
            "텍스트": 결과["텍스트"],
            "완료": 결과 ["완료"]
        }
        결과의 경우
    ]
    JSON 응답(콘텐츠)을 반환합니다.

Async def add_note(요청):
    데이터 = 요청을 기다립니다.json ()
    쿼리 = notes. 삽입().값(
       텍스트 = 데이터 ["텍스트"],
       완료 = 데이터 ["완료"]
    )
    데이터베이스 실행 대기 중 (쿼리)
    JSON 응답을 반환합니다({
        "텍스트": 데이터["텍스트"],
        "완료": 데이터 ["완료"]
    })

경로 = [
    경로("/", endpoint = list_, = ["GET"]),
    경로("/", endpoint = add_note, = ["POST"]),
]

앱 = Starlette (
    경로 = 경로,
    수명 = 수명,
)

마지막으로 데이터 데이터베이스와 테이블을 만들어야 합니다. 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)

거래; 사업; 거래 (이 단어는 여러 의미를 가지며, 구체적인 번역은 문맥에 따라 다릅니다.) 주어진 단어 "거래"만 문자 그대로 번역되면 "거래; 거래"로 번역될 수 있습니다. 다음은 특정 상황에 따라 선택할 수 있는 몇 가지 일반적인 정의입니다. 제목의 요구 사항에 따라 직접 출력은 다음과 같이 변환됩니다.

데이터베이스 트랜잭션은 데코레이터, 컨텍스트 관리자 또는 낮은 수준의 API로 사용할 수 있습니다.

엔드포인트에서 데코레이터 사용:

@ase. ()
Async def ate_note(요청):
    # 이 데이터베이스 삽입은 트랜잭션 내에서 발생합니다.
    #`RuntimeError` 에 의해 롤백됩니다.
    쿼리 = 노트. 삽입 ().값 (텍스트 = "당신은 나를 볼 수 없습니다", 완료 = 참)
    데이터베이스 실행 대기 중 (쿼리)
    런타임 오류 올리기()

컨텍스트 관리자 사용:

Async def ate_note(요청):
    데이터베이스 트랜잭션과 비동기화():
        # 이 데이터베이스 삽입은 트랜잭션 내에서 발생합니다.
        #`RuntimeError` 에 의해 롤백됩니다.
        쿼리 = 노트. 삽입 ().값 (텍스트 = "당신은 나를 볼 수 없습니다", 완료 = 참)
        대기 중인 요청.데이터베이스.실행(쿼리)
        런타임 오류 올리기()

낮은 수준의 API 사용:

Async def ate_note(요청):
    트랜잭션 = 데이터베이스를 기다립니다. 트랜잭션()
    시도:
        # 이 데이터베이스 삽입은 트랜잭션 내에서 발생합니다.
        #`RuntimeError` 에 의해 롤백됩니다.
        쿼리 = 노트. 삽입 ().값 (텍스트 = "당신은 나를 볼 수 없습니다", 완료 = 참)
        데이터베이스 실행 대기 중 (쿼리)
        런타임 오류 올리기()
    제외:
        거래 대기 중. 롤백 ()
        키우기
    기타:
        거래 대기 중. 커밋()

테스트 격리

데이터베이스를 사용하는 서비스에 대해 테스트할 때 몇 가지를 보장해야 합니다. 우리의 요구 사항은 다음과 같아야 합니다.

  • 별도의 데이터베이스로 테스트합니다.
  • 테스트를 실행할 때마다 새 테스트 데이터베이스가 생성됩니다.
  • 각 테스트 케이스 사이의 데이터베이스 상태가 서로 분리되어 있는지 확인합니다.

이러한 요구 사항을 충족하려면 다음과 같은 애플리케이션 및 테스트를 구축해야 합니다.

Starlette. 응용 프로그램이 Starlette에서 가져옵니다.
Starlette.config에서 가져오기 구성
데이터베이스 가져오기

구성 = 구성(. "env")

테스트 = 구성('테스트', 캐스트 = 볼, 기본값 = 거짓)
DATABASE_URL = 구성('DATABASE_URL', 캐스트 = 데이터베이스. aseURL)
TEST_DATABASE_URL = DATABASE_URL (데이터베이스 = 'test_' + DATABASE_URL 데이터베이스)

# 테스트 중에 'force_roll back'을 사용하여 데이터베이스 변경을 지속하지 않도록 합니다.
# 각 테스트 케이스 사이.
테스트하는 경우:
    데이터베이스 = 데이터베이스. 데이터베이스(TEST_DATABASE_URL, force_roll back = 참)
기타:
    데이터베이스 = 데이터베이스. 데이터베이스(DATABASE_URL)

연습 실행 중에 TESTING 를 설정하고 테스트 데이터베이스를 설정해야 합니다.py.test 를 사용한다고 가정하면 conftest.py 는 다음과 같습니다.

파이테스트 가져오기
Starlette.config에서 가져오기 환경
Starlette.testClient에서 TestClient 가져오기
sqlalchemy에서 _
sqlalcheme_utils에서 ase_, create_, drop_를 가져옵니다.

# 이것은 `os.environ` 를 설정하지만 약간의 추가 보호를 제공합니다.
# 애플리케이션 가져오기 아래에 놓으면 오류가 발생합니다.
# '테스트'가 이미 환경에서 읽혔음을 알려줍니다.
환경 ['테스트'] = '진실'

앱 가져오기


@Pytest.ture (범위 = "세션", 자동 사용 = 참)
Def create_test_ ():
  """
  모든 테스트 케이스에 깨끗한 데이터베이스를 만듭니다.
  안전을 위해 데이터베이스가 이미 있는 경우 중단해야 합니다.

  우리는 몇 명의 도우미를 위해 여기에서 `sqlalchemy_utils` 패키지를 지속적으로 사용합니다.
  데이터베이스를 만들고 삭제합니다.
  """
  URL = str (앱. TEST_DATABASE_URL)
  엔진 = create_ (url)
  데이터베이스가 존재하지 않는다고 주장합니다. (url), '테스트 데이터베이스가 이미 존재합니다. 테스트 중단.'
  create_ (url) # 테스트 데이터베이스를 만듭니다.
  메타데이터. create_all (엔진) # 테이블을 만듭니다.
  항복 # 테스트를 실행합니다.
  drop_ (url) # 테스트 데이터베이스를 삭제합니다.


@pytest.ture ()
데프 클라이언트():
    """
    테스트 케이스에서 '클라이언트' 고정장치를 사용하면 전체 데이터베이스를 얻을 수 있습니다.
    테스트 사례 간 롤백:

    Def test_ (클라이언트):
        URL = app.url_pase_for ('홈페이지')
        응답 = client.get (url)
        응답을 주장합니다. 상태 _ 코드 = = 200
    """
    TestClient(앱)를 클라이언트로 사용:
        양보 클라이언트

마이그레이션

데이터베이스의 증분 변경을 관리하려면 데이터베이스 마이그레이션을 사용해야 합니다. 이를 위해 SQLAlchemy의 저자가 쓴 Alembic을 적극 권장합니다.

$Pip 설치 alembic
$alembic init 마이그레이션

이제 Alembic이 구성된 DATABASE_URL을 참조하고 테이블 메타데이터를 사용하도록 설정해야 합니다.

alembic.ini 에서 다음 줄을 삭제합니다.

Sqlalchey.url = 드라이버: // 사용자: pass@localhost/dbname

migrations/env.py 에서는 'sqlalchemy.url' 구성 키와 target_metadata 변수를 설정해야 합니다. 다음과 같은 것이 필요합니다.

# Alembic Config 개체.
컨텍스트 = 컨텍스트.config

# DATABASE_URL 및 테이블 정의를 사용하도록 Alembic을 구성합니다. ..
앱 가져오기
ConFig.set_main _ 옵션('sqlalchey.url', str(app.DATABASE_URL))
target_metadata = app.metadata

...

그런 다음 위의 노트 예제를 사용하여 예비 수정본을 작성합니다.

Alembic -m "노트 만들기 테이블"

그리고 필요한 지침으로 새 파일(migrations/versions 내부)을 채웁니다.

데프 업그레이드():
    Op. create_table (
      '노트',
      Sqlalchemy. 열("id", sqlalchemy. 정수, prial_key = 참),
      Sqlalchemy. 열("텍스트", sqlalchemy. 문자열),
      Sqlalchemy. 열("완료", sqlalchemy. 부울),
    )

데프 다운그레이드():
    Op. drop_table('')

그리고 첫 번째 마이그레이션을 실행합니다. 노트 필기 앱이 지금 실행 중입니다!

알렘빅 업그레이드 헤드

테스트 중 마이그레이션 실행

테스트 데이터베이스를 만들 때마다 테스트 제품군이 데이터베이스 마이그레이션을 실행하도록 하는 것이 좋습니다. 이렇게 하면 마이그레이션 스크립트의 문제를 식별하고 실시간 데이터베이스 상태와 일치하는 데이터베이스에 대해 테스트가 실행되도록 하는 데 도움이 됩니다.

create_test_database 고정장치를 약간 조정할 수 있습니다.

alembic 가져오기 명령에서
alembig.config에서 가져오기 구성
앱 가져오기

...

@Pytest.ture (범위 = "세션", 자동 사용 = 참)
Def create_test_ ():
    URL = str (앱. DATABASE_URL)
    엔진 = create_ (url)
    데이터베이스가 존재하지 않는다고 주장합니다. (url), '테스트 데이터베이스가 이미 존재합니다. 테스트 중단.'
    create_ (url) # 테스트 데이터베이스를 만듭니다.
    Config = Config("alemvic.ini") # 마이그레이션을 실행합니다.
    명령.업그레이드(config, "head")
    항복 # 테스트를 실행합니다.
    drop_ (url) # 테스트 데이터베이스를 삭제합니다.