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 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 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 contextvars import ContextVar

import peewee

DATABASE_NAME = "test.db"
db_state_default = {"closed": None, "conn": None, "ctx": None, "transactions": None}
db_state = ContextVar("db_state", default=db_state_default.copy())


class PeeweeConnectionState(peewee._ConnectionState):
    def __init__(self, **kwargs):
        super().__setattr__("_state", db_state)
        super().__init__(**kwargs)

    def __setattr__(self, name, value):

👉 🖼 👥 🔜 ✍ 🍕 💽 🕴 💯.

## 📁 📊

👥 ✍ 🆕 📁 `sql_app/tests/test_sql_app.py`.

🆕 📁 📊 👀 💖:

``` hl_lines="9-11"
.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    ├── schemas.py
    └── tests
        ├── __init__.py
        └── test_sql_app.py
```

## ✍ 🆕 💽 🎉

🥇, 👥 ✍ 🆕 💽 🎉 ⏮️ 🆕 💽.

测试时使用覆盖依赖项返回自定义数据库会话代替正常的依赖项。

本例中，要创建仅用于测试的临时数据库。

## 文件架构

创建新文件 `sql_app/tests/test_sql_app.py`。

因此，新的文件架构如下：

``` hl_lines="9-11"
.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    ├── schemas.py
    └── tests
        ├── __init__.py
        └── test_sql_app.py
```

## 创建新的数据库会话

首先，为新建数据库创建新的数据库会话。

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    ├── schemas.py
    └── tests
        ├── __init__.py
        └── test_sql_app.py
```

## Create the new database session

First, we create a new database session with the new database.

We'll use an in-memory database that persists during the tests instead of the local file `sql_app.db`.

from fastapi import Depends, FastAPI, HTTPException

from . import crud, database, models, schemas
from .database import db_state_default

database.db.connect()
database.db.create_tables([models.User, models.Item])
database.db.close()

app = FastAPI()

sleep_time = 10


async def reset_db_state():
    database.db._state._state.set(db_state_default.copy())
    database.db._state.reset()

initial_fake_database = {"rick": "Rick Sanchez"}

fake_database = initial_fake_database.copy()

initial_state = {"except": False, "finally": False}

state = initial_state.copy()

app = FastAPI()


async def get_database():
    temp_database = fake_database.copy()
    try:
        yield temp_database
        fake_database.update(temp_database)
    except HTTPException:

pkg database/sql, const LevelDefault = 0
pkg database/sql, const LevelDefault IsolationLevel
pkg database/sql, const LevelLinearizable = 7
pkg database/sql, const LevelLinearizable IsolationLevel
pkg database/sql, const LevelReadCommitted = 2
pkg database/sql, const LevelReadCommitted IsolationLevel
pkg database/sql, const LevelReadUncommitted = 1
pkg database/sql, const LevelReadUncommitted IsolationLevel