# 异步 SQL 关系型数据库

**FastAPI** 使用 <a href="https://github.com/encode/databases" class="external-link" target="_blank">`encode/databases`</a> 为连接数据库提供异步支持（`async` 与 `await`）。

`databases` 兼容以下数据库：

* PostgreSQL
* MySQL
* SQLite

本章示例使用 **SQLite**，它使用的是单文件，且 Python 内置集成了 SQLite，因此，可以直接复制并运行本章示例。

生产环境下，则要使用 **PostgreSQL** 等数据库服务器。

!!! tip "提示"

from ...utils import needs_pydanticv1


@pytest.fixture(scope="module")
def client():
    test_db = Path("./sql_app.db")
    if test_db.is_file():  # pragma: nocover
        test_db.unlink()
    # Import while creating the client to create the DB after starting the test session
    from docs_src.sql_databases.sql_app import alt_main

    # Ensure import side effects are re-executed
    importlib.reload(alt_main)

    tmp_path = tmp_path_factory.mktemp("data")
    cwd = os.getcwd()
    os.chdir(tmp_path)
    test_db = Path("./sql_app.db")
    if test_db.is_file():  # pragma: nocover
        test_db.unlink()
    # Import while creating the client to create the DB after starting the test session
    from docs_src.sql_databases.sql_app_py310 import alt_main

    # Ensure import side effects are re-executed
    importlib.reload(alt_main)

Harsha Laxman <******@****.***> 1687432812 -0700

    tmp_path = tmp_path_factory.mktemp("data")
    cwd = os.getcwd()
    os.chdir(tmp_path)
    test_db = Path("./sql_app.db")
    if test_db.is_file():  # pragma: nocover
        test_db.unlink()
    # Import while creating the client to create the DB after starting the test session
    from docs_src.sql_databases.sql_app_py310 import main

    # Ensure import side effects are re-executed
    importlib.reload(main)

    tmp_path = tmp_path_factory.mktemp("data")
    cwd = os.getcwd()
    os.chdir(tmp_path)
    test_db = Path("./sql_app.db")
    if test_db.is_file():  # pragma: nocover
        test_db.unlink()
    # Import while creating the client to create the DB after starting the test session
    from docs_src.sql_databases.sql_app_py39 import main

    # Ensure import side effects are re-executed
    importlib.reload(main)

    tmp_path = tmp_path_factory.mktemp("data")
    cwd = os.getcwd()
    os.chdir(tmp_path)
    test_db = Path("./sql_app.db")
    if test_db.is_file():  # pragma: nocover
        test_db.unlink()
    # Import while creating the client to create the DB after starting the test session
    from docs_src.sql_databases.sql_app_py39 import alt_main

    # Ensure import side effects are re-executed
    importlib.reload(alt_main)

{!../../../docs_src/sql_databases/sql_app/tests/test_sql_app.py!}
```

!!! tip
    The code for `override_get_db()` is almost exactly the same as for `get_db()`, but in `override_get_db()` we use the `TestingSessionLocal` for the testing database instead.

## Test the app

Then we can just test the app as normally.

```Python hl_lines="32-47"
{!../../../docs_src/sql_databases/sql_app/tests/test_sql_app.py!}
```

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from fastapi.testclient import TestClient
from sqlalchemy.sql.elements import quoted_name

app = FastAPI(default_response_class=ORJSONResponse)


@app.get("/orjson_non_str_keys")
def get_orjson_non_str_keys():
    key = quoted_name(value="msg", quote=False)
    return {key: "Hello World", 1: 1}


client = TestClient(app)


def test_orjson_non_str_keys():