Now we will see how useful it is to have this session dependency. ✨
-{* ./docs_src/tutorial/fastapi/app_testing/tutorial001/main.py ln[0] *}
+{* ./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/main.py ln[0] *}
## File Structure
Let's start with a simple test, with just the basic test code we need the check that the **FastAPI** application is creating a new hero correctly.
```{ .python .annotate }
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_001.py[ln:1-7]!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_001.py[ln:1-7]!}
# Some code here omitted, we will see it later 👈
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_001.py[ln:20-24]!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_001.py[ln:20-24]!}
# Some code here omitted, we will see it later 👈
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_001.py[ln:26-32]!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_001.py[ln:26-32]!}
# Code below omitted 👇
```
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/annotations/en/test_main_001.md!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/annotations/en/test_main_001.md!}
/// tip
That way we protect the production database and we have better control of the data we are testing.
```{ .python .annotate hl_lines="4 9-10 12 19" }
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_002.py[ln:1-7]!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_002.py[ln:1-7]!}
# Some code here omitted, we will see it later 👈
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_002.py[ln:15-32]!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_002.py[ln:15-32]!}
# Code below omitted 👇
```
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/annotations/en/test_main_002.md!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/annotations/en/test_main_002.md!}
/// tip
So, the testing database will be in the file `testing.db`.
``` { .python .annotate hl_lines="4 8-11 13 16 33"}
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_003.py!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_003.py!}
```
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/annotations/en/test_main_003.md!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/annotations/en/test_main_003.md!}
### Import Table Models
We just have to change a couple of parameters in the **engine**.
```{ .python .annotate hl_lines="3 9-13"}
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_004.py[ln:1-13]!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_004.py[ln:1-13]!}
# Code below omitted 👇
```
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/annotations/en/test_main_004.md!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/annotations/en/test_main_004.md!}
/// tip
Let's see the first code example with a fixture:
``` { .python .annotate }
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_005.py!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_005.py!}
```
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/annotations/en/test_main_005.md!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/annotations/en/test_main_005.md!}
/// tip
So, we can create a **client fixture** that will be used in all the tests, and it will itself require the **session fixture**.
``` { .python .annotate hl_lines="19-28 31" }
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main_006.py!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main_006.py!}
```
-{!./docs_src/tutorial/fastapi/app_testing/tutorial001/annotations/en/test_main_006.md!}
+{!./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/annotations/en/test_main_006.md!}
/// tip
Let's add some more tests:
-{* ./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main.py ln[30:58] hl[30,49] *}
+{* ./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main.py ln[30:58] hl[30,49] *}
/// tip
But for the next test function, we will require **both fixtures**, the **client** and the **session**.
-{* ./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main.py ln[1:6,61:81] hl[6,61] *}
+{* ./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main.py ln[1:6,61:81] hl[6,61] *}
In this test function, we want to check that the *path operation* to **read a list of heroes** actually sends us heroes.
Using the same ideas, requiring the fixtures, creating data that we need for the tests, etc., we can now add the rest of the tests. They look quite similar to what we have done up to now.
-{* ./docs_src/tutorial/fastapi/app_testing/tutorial001/test_main.py ln[84:125] hl[84,99,114] *}
+{* ./docs_src/tutorial/fastapi/app_testing/tutorial001_py310/test_main.py ln[84:125] hl[84,99,114] *}
## Run the Tests
+++ /dev/null
-1. Import the `app` from the the `main` module.
-
-2. We create a `TestClient` for the FastAPI `app` and put it in the variable `client`.
-
-3. Then we use use this `client` to **talk to the API** and send a `POST` HTTP operation, creating a new hero.
-
-4. Then we get the **JSON data** from the response and put it in the variable `data`.
-
-5. Next we start testing the results with `assert` statements, we check that the status code of the response is `200`.
-
-6. We check that the `name` of the hero created is `"Deadpond"`.
-
-7. We check that the `secret_name` of the hero created is `"Dive Wilson"`.
-
-8. We check that the `age` of the hero created is `None`, because we didn't send an age.
-
-9. We check that the hero created has an `id` created by the database, so it's not `None`.
+++ /dev/null
-1. Import the `get_session` dependency from the the `main` module.
-
-2. Define the new function that will be the new **dependency override**.
-
-3. This function will return a different **session** than the one that would be returned by the original `get_session` function.
-
- We haven't seen how this new **session** object is created yet, but the point is that this is a different session than the original one from the app.
-
- This session is attached to a different **engine**, and that different **engine** uses a different URL, for a database just for testing.
-
- We haven't defined that new **URL** nor the new **engine** yet, but here we already see the that this object `session` will override the one returned by the original dependency `get_session()`.
-
-4. Then, the FastAPI `app` object has an attribute `app.dependency_overrides`.
-
- This attribute is a dictionary, and we can put dependency overrides in it by passing, as the **key**, the **original dependency function**, and as the **value**, the **new overriding dependency function**.
-
- So, here we are telling the FastAPI app to use `get_session_override` instead of `get_session` in all the places in the code that depend on `get_session`, that is, all the parameters with something like:
-
- ```Python
- session: Session = Depends(get_session)
- ```
-
-5. After we are done with the dependency override, we can restore the application back to normal, by removing all the values in this dictionary `app.dependency_overrides`.
-
- This way whenever a *path operation function* needs the dependency FastAPI will use the original one instead of the override.
+++ /dev/null
-1. Here's a subtle thing to notice.
-
- Remember that [Order Matters](../create-db-and-table.md#sqlmodel-metadata-order-matters){.internal-link target=_blank} and we need to make sure all the **SQLModel** models are already defined and **imported** before calling `.create_all()`.
-
- IN this line, by importing something, *anything*, from `.main`, the code in `.main` will be executed, including the definition of the **table models**, and that will automatically register them in `SQLModel.metadata`.
-
-2. Here we create a new **engine**, completely different from the one in `main.py`.
-
- This is the engine we will use for the tests.
-
- We use the new URL of the database for tests:
-
- ```
- sqlite:///testing.db
- ```
-
- And again, we use the connection argument `check_same_thread=False`.
-
-3. Then we call:
-
- ```Python
- SQLModel.metadata.create_all(engine)
- ```
-
- ...to make sure we create all the tables in the new testing database.
-
- The **table models** are registered in `SQLModel.metadata` just because we imported *something* from `.main`, and the code in `.main` was executed, creating the classes for the **table models** and automatically registering them in `SQLModel.metadata`.
-
- So, by the point we call this method, the **table models** are already registered there. 💯
-
-4. Here's where we create the custom **session** object for this test in a `with` block.
-
- It uses the new custom **engine** we created, so anything that uses this session will be using the testing database.
-
-5. Now, back to the dependency override, it is just returning the same **session** object from outside, that's it, that's the whole trick.
-
-6. By this point, the testing **session** `with` block finishes, and the session is closed, the file is closed, etc.
+++ /dev/null
-1. Import `StaticPool` from `sqlmodel`, we will use it in a bit.
-
-2. For the **SQLite URL**, don't write any file name, leave it empty.
-
- So, instead of:
-
- ```
- sqlite:///testing.db
- ```
-
- ...just write:
-
- ```
- sqlite://
- ```
-
- This is enough to tell **SQLModel** (actually SQLAlchemy) that we want to use an **in-memory SQLite database**.
-
-3. Remember that we told the **low-level** library in charge of communicating with SQLite that we want to be able to **access the database from different threads** with `check_same_thread=False`?
-
- Now that we use an **in-memory database**, we need to also tell SQLAlchemy that we want to be able to use the **same in-memory database** object from different threads.
-
- We tell it that with the `poolclass=StaticPool` parameter.
-
- /// info
-
- You can read more details in the <a href="https://docs.sqlalchemy.org/en/14/dialects/sqlite.html#using-a-memory-database-in-multiple-threads" class="external-link" target="_blank">SQLAlchemy documentation about Using a Memory Database in Multiple Threads</a>
-
- ///
+++ /dev/null
-1. Import `pytest`.
-
-2. Use the `@pytest.fixture()` decorator on top of the function to tell pytest that this is a **fixture** function (equivalent to a FastAPI dependency).
-
- We also give it a name of `"session"`, this will be important in the testing function.
-
-3. Create the fixture function. This is equivalent to a FastAPI dependency function.
-
- In this fixture we create the custom **engine**, with the in-memory database, we create the tables, and we create the **session**.
-
- Then we `yield` the `session` object.
-
-4. The thing that we `return` or `yield` is what will be available to the test function, in this case, the `session` object.
-
- Here we use `yield` so that **pytest** comes back to execute "the rest of the code" in this function once the testing function is done.
-
- We don't have any more visible "rest of the code" after the `yield`, but we have the end of the `with` block that will close the **session**.
-
- By using `yield`, pytest will:
-
- * run the first part
- * create the **session** object
- * give it to the test function
- * run the test function
- * once the test function is done, it will continue here, right after the `yield`, and will correctly close the **session** object in the end of the `with` block.
-
-5. Now, in the test function, to tell **pytest** that this test wants to get the fixture, instead of declaring something like in FastAPI with:
-
- ```Python
- session: Session = Depends(session_fixture)
- ```
-
- ...the way we tell pytest what is the fixture that we want is by using the **exact same name** of the fixture.
-
- In this case, we named it `session`, so the parameter has to be exactly named `session` for it to work.
-
- We also add the type annotation `session: Session` so that we can get autocompletion and inline error checks in our editor.
-
-6. Now in the dependency override function, we just return the same `session` object that came from outside it.
-
- The `session` object comes from the parameter passed to the test function, and we just re-use it and return it here in the dependency override.
+++ /dev/null
-1. Create the new fixture named `"client"`.
-
-2. This **client fixture**, in turn, also requires the **session fixture**.
-
-3. Now we create the **dependency override** inside the client fixture.
-
-4. Set the **dependency override** in the `app.dependency_overrides` dictionary.
-
-5. Create the `TestClient` with the **FastAPI** `app`.
-
-6. `yield` the `TestClient` instance.
-
- By using `yield`, after the test function is done, pytest will come back to execute the rest of the code after `yield`.
-
-7. This is the cleanup code, after `yield`, and after the test function is done.
-
- Here we clear the dependency overrides (here it's only one) in the FastAPI `app`.
-
-8. Now the test function requires the **client fixture**.
-
- And inside the test function, the code is quite **simple**, we just use the `TestClient` to make requests to the API, check the data, and that's it.
-
- The fixtures take care of all the **setup** and **cleanup** code.
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import Depends, FastAPI, HTTPException, Query
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def get_session():
- with Session(engine) as session:
- yield session
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(
- *,
- session: Session = Depends(get_session),
- offset: int = 0,
- limit: int = Query(default=100, le=100),
-):
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(
- *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
-):
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- db_hero.sqlmodel_update(hero_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.delete("/heroes/{hero_id}")
-def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- session.delete(hero)
- session.commit()
- return {"ok": True}
+++ /dev/null
-from fastapi.testclient import TestClient
-from sqlalchemy import Inspector, inspect
-from sqlmodel import Session, create_engine
-
-from . import main as app_mod
-from .test_main import client_fixture, session_fixture
-
-assert client_fixture, "This keeps the client fixture used below"
-assert session_fixture, "This keeps the session fixture used by client_fixture"
-
-
-def test_startup():
- app_mod.engine = create_engine("sqlite://")
- app_mod.on_startup()
- insp: Inspector = inspect(app_mod.engine)
- assert insp.has_table(str(app_mod.Hero.__tablename__))
-
-
-def test_get_session():
- app_mod.engine = create_engine("sqlite://")
- for session in app_mod.get_session():
- assert isinstance(session, Session)
- assert session.bind == app_mod.engine
-
-
-def test_read_hero_not_found(client: TestClient):
- response = client.get("/heroes/9000")
- assert response.status_code == 404
-
-
-def test_update_hero_not_found(client: TestClient):
- response = client.patch("/heroes/9000", json={"name": "Very-Rusty-Man"})
- assert response.status_code == 404
-
-
-def test_delete_hero_not_found(client: TestClient):
- response = client.delete("/heroes/9000")
- assert response.status_code == 404
+++ /dev/null
-import pytest
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-from sqlmodel.pool import StaticPool
-
-from .main import Hero, app, get_session
-
-
-@pytest.fixture(name="session")
-def session_fixture():
- engine = create_engine(
- "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
- )
- SQLModel.metadata.create_all(engine)
- with Session(engine) as session:
- yield session
-
-
-@pytest.fixture(name="client")
-def client_fixture(session: Session):
- def get_session_override():
- return session
-
- app.dependency_overrides[get_session] = get_session_override
- client = TestClient(app)
- yield client
- app.dependency_overrides.clear()
-
-
-def test_create_hero(client: TestClient):
- response = client.post(
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpond"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] is not None
-
-
-def test_create_hero_incomplete(client: TestClient):
- # No secret_name
- response = client.post("/heroes/", json={"name": "Deadpond"})
- assert response.status_code == 422
-
-
-def test_create_hero_invalid(client: TestClient):
- # secret_name has an invalid type
- response = client.post(
- "/heroes/",
- json={
- "name": "Deadpond",
- "secret_name": {"message": "Do you wanna know my secret identity?"},
- },
- )
- assert response.status_code == 422
-
-
-def test_read_heroes(session: Session, client: TestClient):
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
- hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
- session.add(hero_1)
- session.add(hero_2)
- session.commit()
-
- response = client.get("/heroes/")
- data = response.json()
-
- assert response.status_code == 200
-
- assert len(data) == 2
- assert data[0]["name"] == hero_1.name
- assert data[0]["secret_name"] == hero_1.secret_name
- assert data[0]["age"] == hero_1.age
- assert data[0]["id"] == hero_1.id
- assert data[1]["name"] == hero_2.name
- assert data[1]["secret_name"] == hero_2.secret_name
- assert data[1]["age"] == hero_2.age
- assert data[1]["id"] == hero_2.id
-
-
-def test_read_hero(session: Session, client: TestClient):
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
- session.add(hero_1)
- session.commit()
-
- response = client.get(f"/heroes/{hero_1.id}")
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == hero_1.name
- assert data["secret_name"] == hero_1.secret_name
- assert data["age"] == hero_1.age
- assert data["id"] == hero_1.id
-
-
-def test_update_hero(session: Session, client: TestClient):
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
- session.add(hero_1)
- session.commit()
-
- response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpuddle"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] == hero_1.id
-
-
-def test_delete_hero(session: Session, client: TestClient):
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
- session.add(hero_1)
- session.commit()
-
- response = client.delete(f"/heroes/{hero_1.id}")
-
- hero_in_db = session.get(Hero, hero_1.id)
-
- assert response.status_code == 200
-
- assert hero_in_db is None
+++ /dev/null
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-
-from .main import app, get_session # (1)!
-
-
-def test_create_hero():
- engine = create_engine(
- "sqlite:///testing.db", connect_args={"check_same_thread": False}
- )
- SQLModel.metadata.create_all(engine)
-
- with Session(engine) as session:
-
- def get_session_override():
- return session
-
- app.dependency_overrides[get_session] = get_session_override
-
- client = TestClient(app) # (2)!
-
- response = client.post( # (3)!
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- app.dependency_overrides.clear()
- data = response.json() # (4)!
-
- assert response.status_code == 200 # (5)!
- assert data["name"] == "Deadpond" # (6)!
- assert data["secret_name"] == "Dive Wilson" # (7)!
- assert data["age"] is None # (8)!
- assert data["id"] is not None # (9)!
+++ /dev/null
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-
-from .main import app, get_session # (1)!
-
-
-def test_create_hero():
- engine = create_engine(
- "sqlite:///testing.db", connect_args={"check_same_thread": False}
- )
- SQLModel.metadata.create_all(engine)
-
- with Session(engine) as session:
-
- def get_session_override(): # (2)!
- return session # (3)!
-
- app.dependency_overrides[get_session] = get_session_override # (4)!
-
- client = TestClient(app)
-
- response = client.post(
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- app.dependency_overrides.clear() # (5)!
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpond"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] is not None
+++ /dev/null
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-
-from .main import app, get_session # (1)!
-
-
-def test_create_hero():
- engine = create_engine( # (2)!
- "sqlite:///testing.db", connect_args={"check_same_thread": False}
- )
- SQLModel.metadata.create_all(engine) # (3)!
-
- with Session(engine) as session: # (4)!
-
- def get_session_override():
- return session # (5)!
-
- app.dependency_overrides[get_session] = get_session_override # (4)!
-
- client = TestClient(app)
-
- response = client.post(
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- app.dependency_overrides.clear()
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpond"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] is not None
- # (6)!
+++ /dev/null
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-from sqlmodel.pool import StaticPool # (1)!
-
-from .main import app, get_session
-
-
-def test_create_hero():
- engine = create_engine(
- "sqlite://", # (2)!
- connect_args={"check_same_thread": False},
- poolclass=StaticPool, # (3)!
- )
- SQLModel.metadata.create_all(engine)
-
- with Session(engine) as session:
-
- def get_session_override():
- return session
-
- app.dependency_overrides[get_session] = get_session_override
-
- client = TestClient(app)
-
- response = client.post(
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- app.dependency_overrides.clear()
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpond"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] is not None
+++ /dev/null
-import pytest # (1)!
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-from sqlmodel.pool import StaticPool
-
-from .main import app, get_session
-
-
-@pytest.fixture(name="session") # (2)!
-def session_fixture(): # (3)!
- engine = create_engine(
- "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
- )
- SQLModel.metadata.create_all(engine)
- with Session(engine) as session:
- yield session # (4)!
-
-
-def test_create_hero(session: Session): # (5)!
- def get_session_override():
- return session # (6)!
-
- app.dependency_overrides[get_session] = get_session_override
-
- client = TestClient(app)
-
- response = client.post(
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- app.dependency_overrides.clear()
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpond"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] is not None
+++ /dev/null
-import pytest
-from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine
-from sqlmodel.pool import StaticPool
-
-from .main import app, get_session
-
-
-@pytest.fixture(name="session")
-def session_fixture():
- engine = create_engine(
- "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
- )
- SQLModel.metadata.create_all(engine)
- with Session(engine) as session:
- yield session
-
-
-@pytest.fixture(name="client") # (1)!
-def client_fixture(session: Session): # (2)!
- def get_session_override(): # (3)!
- return session
-
- app.dependency_overrides[get_session] = get_session_override # (4)!
-
- client = TestClient(app) # (5)!
- yield client # (6)!
- app.dependency_overrides.clear() # (7)!
-
-
-def test_create_hero(client: TestClient): # (8)!
- response = client.post(
- "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
- )
- data = response.json()
-
- assert response.status_code == 200
- assert data["name"] == "Deadpond"
- assert data["secret_name"] == "Dive Wilson"
- assert data["age"] is None
- assert data["id"] is not None
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI, HTTPException, Query
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- with Session(engine) as session:
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
- with Session(engine) as session:
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(hero_id: int):
- with Session(engine) as session:
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(hero_id: int, hero: HeroUpdate):
- with Session(engine) as session:
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- db_hero.sqlmodel_update(hero_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.delete("/heroes/{hero_id}")
-def delete_hero(hero_id: int):
- with Session(engine) as session:
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- session.delete(hero)
- session.commit()
- return {"ok": True}
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI, HTTPException, Query
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- with Session(engine) as session:
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
- with Session(engine) as session:
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(hero_id: int):
- with Session(engine) as session:
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class Hero(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class HeroCreate(SQLModel):
- name: str
- secret_name: str
- age: Optional[int] = None
-
-
-class HeroPublic(SQLModel):
- id: int
- name: str
- secret_name: str
- age: Optional[int] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- with Session(engine) as session:
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes():
- with Session(engine) as session:
- heroes = session.exec(select(Hero)).all()
- return heroes
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- with Session(engine) as session:
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes():
- with Session(engine) as session:
- heroes = session.exec(select(Hero)).all()
- return heroes
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI, HTTPException
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- with Session(engine) as session:
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes():
- with Session(engine) as session:
- heroes = session.exec(select(Hero)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(hero_id: int):
- with Session(engine) as session:
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import Depends, FastAPI, HTTPException, Query
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class TeamBase(SQLModel):
- name: str = Field(index=True)
- headquarters: str
-
-
-class Team(TeamBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-class TeamCreate(TeamBase):
- pass
-
-
-class TeamPublic(TeamBase):
- id: int
-
-
-class TeamUpdate(SQLModel):
- id: Optional[int] = None
- name: Optional[str] = None
- headquarters: Optional[str] = None
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
- team_id: Optional[int] = None
-
-
-class HeroPublicWithTeam(HeroPublic):
- team: Optional[TeamPublic] = None
-
-
-class TeamPublicWithHeroes(TeamPublic):
- heroes: List[HeroPublic] = []
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def get_session():
- with Session(engine) as session:
- yield session
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(
- *,
- session: Session = Depends(get_session),
- offset: int = 0,
- limit: int = Query(default=100, le=100),
-):
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublicWithTeam)
-def read_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(
- *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
-):
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- db_hero.sqlmodel_update(hero_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.delete("/heroes/{hero_id}")
-def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- session.delete(hero)
- session.commit()
- return {"ok": True}
-
-
-@app.post("/teams/", response_model=TeamPublic)
-def create_team(*, session: Session = Depends(get_session), team: TeamCreate):
- db_team = Team.model_validate(team)
- session.add(db_team)
- session.commit()
- session.refresh(db_team)
- return db_team
-
-
-@app.get("/teams/", response_model=List[TeamPublic])
-def read_teams(
- *,
- session: Session = Depends(get_session),
- offset: int = 0,
- limit: int = Query(default=100, le=100),
-):
- teams = session.exec(select(Team).offset(offset).limit(limit)).all()
- return teams
-
-
-@app.get("/teams/{team_id}", response_model=TeamPublicWithHeroes)
-def read_team(*, team_id: int, session: Session = Depends(get_session)):
- team = session.get(Team, team_id)
- if not team:
- raise HTTPException(status_code=404, detail="Team not found")
- return team
-
-
-@app.patch("/teams/{team_id}", response_model=TeamPublic)
-def update_team(
- *,
- session: Session = Depends(get_session),
- team_id: int,
- team: TeamUpdate,
-):
- db_team = session.get(Team, team_id)
- if not db_team:
- raise HTTPException(status_code=404, detail="Team not found")
- team_data = team.model_dump(exclude_unset=True)
- db_team.sqlmodel_update(team_data)
- session.add(db_team)
- session.commit()
- session.refresh(db_team)
- return db_team
-
-
-@app.delete("/teams/{team_id}")
-def delete_team(*, session: Session = Depends(get_session), team_id: int):
- team = session.get(Team, team_id)
- if not team:
- raise HTTPException(status_code=404, detail="Team not found")
- session.delete(team)
- session.commit()
- return {"ok": True}
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class Hero(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=Hero)
-def create_hero(hero: Hero):
- with Session(engine) as session:
- session.add(hero)
- session.commit()
- session.refresh(hero)
- return hero
-
-
-@app.get("/heroes/", response_model=List[Hero])
-def read_heroes():
- with Session(engine) as session:
- heroes = session.exec(select(Hero)).all()
- return heroes
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import Depends, FastAPI, HTTPException, Query
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def get_session():
- with Session(engine) as session:
- yield session
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(
- *,
- session: Session = Depends(get_session),
- offset: int = 0,
- limit: int = Query(default=100, le=100),
-):
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(
- *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
-):
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- db_hero.sqlmodel_update(hero_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.delete("/heroes/{hero_id}")
-def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- session.delete(hero)
- session.commit()
- return {"ok": True}
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import Depends, FastAPI, HTTPException, Query
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class TeamBase(SQLModel):
- name: str = Field(index=True)
- headquarters: str
-
-
-class Team(TeamBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-class TeamCreate(TeamBase):
- pass
-
-
-class TeamPublic(TeamBase):
- id: int
-
-
-class TeamUpdate(SQLModel):
- name: Optional[str] = None
- headquarters: Optional[str] = None
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
- team_id: Optional[int] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def get_session():
- with Session(engine) as session:
- yield session
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(
- *,
- session: Session = Depends(get_session),
- offset: int = 0,
- limit: int = Query(default=100, le=100),
-):
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(
- *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
-):
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- db_hero.sqlmodel_update(hero_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.delete("/heroes/{hero_id}")
-def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- session.delete(hero)
- session.commit()
- return {"ok": True}
-
-
-@app.post("/teams/", response_model=TeamPublic)
-def create_team(*, session: Session = Depends(get_session), team: TeamCreate):
- db_team = Team.model_validate(team)
- session.add(db_team)
- session.commit()
- session.refresh(db_team)
- return db_team
-
-
-@app.get("/teams/", response_model=List[TeamPublic])
-def read_teams(
- *,
- session: Session = Depends(get_session),
- offset: int = 0,
- limit: int = Query(default=100, le=100),
-):
- teams = session.exec(select(Team).offset(offset).limit(limit)).all()
- return teams
-
-
-@app.get("/teams/{team_id}", response_model=TeamPublic)
-def read_team(*, team_id: int, session: Session = Depends(get_session)):
- team = session.get(Team, team_id)
- if not team:
- raise HTTPException(status_code=404, detail="Team not found")
- return team
-
-
-@app.patch("/teams/{team_id}", response_model=TeamPublic)
-def update_team(
- *,
- session: Session = Depends(get_session),
- team_id: int,
- team: TeamUpdate,
-):
- db_team = session.get(Team, team_id)
- if not db_team:
- raise HTTPException(status_code=404, detail="Team not found")
- team_data = team.model_dump(exclude_unset=True)
- db_team.sqlmodel_update(team_data)
- session.add(db_team)
- session.commit()
- session.refresh(db_team)
- return db_team
-
-
-@app.delete("/teams/{team_id}")
-def delete_team(*, session: Session = Depends(get_session), team_id: int):
- team = session.get(Team, team_id)
- if not team:
- raise HTTPException(status_code=404, detail="Team not found")
- session.delete(team)
- session.commit()
- return {"ok": True}
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI, HTTPException, Query
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
-
-
-class HeroCreate(HeroBase):
- pass
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- with Session(engine) as session:
- db_hero = Hero.model_validate(hero)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
- with Session(engine) as session:
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(hero_id: int):
- with Session(engine) as session:
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(hero_id: int, hero: HeroUpdate):
- with Session(engine) as session:
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- db_hero.sqlmodel_update(hero_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
+++ /dev/null
-from typing import List, Optional
-
-from fastapi import FastAPI, HTTPException, Query
-from sqlmodel import Field, Session, SQLModel, create_engine, select
-
-
-class HeroBase(SQLModel):
- name: str = Field(index=True)
- secret_name: str
- age: Optional[int] = Field(default=None, index=True)
-
-
-class Hero(HeroBase, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- hashed_password: str = Field()
-
-
-class HeroCreate(HeroBase):
- password: str
-
-
-class HeroPublic(HeroBase):
- id: int
-
-
-class HeroUpdate(SQLModel):
- name: Optional[str] = None
- secret_name: Optional[str] = None
- age: Optional[int] = None
- password: Optional[str] = None
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-connect_args = {"check_same_thread": False}
-engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def hash_password(password: str) -> str:
- # Use something like passlib here
- return f"not really hashed {password} hehehe"
-
-
-app = FastAPI()
-
-
-@app.on_event("startup")
-def on_startup():
- create_db_and_tables()
-
-
-@app.post("/heroes/", response_model=HeroPublic)
-def create_hero(hero: HeroCreate):
- hashed_password = hash_password(hero.password)
- with Session(engine) as session:
- extra_data = {"hashed_password": hashed_password}
- db_hero = Hero.model_validate(hero, update=extra_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
-
-
-@app.get("/heroes/", response_model=List[HeroPublic])
-def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
- with Session(engine) as session:
- heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
- return heroes
-
-
-@app.get("/heroes/{hero_id}", response_model=HeroPublic)
-def read_hero(hero_id: int):
- with Session(engine) as session:
- hero = session.get(Hero, hero_id)
- if not hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- return hero
-
-
-@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
-def update_hero(hero_id: int, hero: HeroUpdate):
- with Session(engine) as session:
- db_hero = session.get(Hero, hero_id)
- if not db_hero:
- raise HTTPException(status_code=404, detail="Hero not found")
- hero_data = hero.model_dump(exclude_unset=True)
- extra_data = {}
- if "password" in hero_data:
- password = hero_data["password"]
- hashed_password = hash_password(password)
- extra_data["hashed_password"] = hashed_password
- db_hero.sqlmodel_update(hero_data, update=extra_data)
- session.add(db_hero)
- session.commit()
- session.refresh(db_hero)
- return db_hero
import pytest
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@dataclass
@pytest.fixture(
name="modules_path",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@dataclass
@pytest.fixture(
name="modules_path",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
app_mod = sys.modules[app_mod_path]
importlib.reload(app_mod)
else:
- app_mod = importlib.import_module(app_mod_path)
+ app_mod = importlib.import_module(app_mod_path) # pragma: no cover
test_mod = importlib.import_module(f"{modules_path}.test_main_002")
return Modules(app=app_mod, test=test_mod)
import pytest
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@dataclass
@pytest.fixture(
name="modules_path",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
app_mod = sys.modules[app_mod_path]
importlib.reload(app_mod)
else:
- app_mod = importlib.import_module(app_mod_path)
+ app_mod = importlib.import_module(app_mod_path) # pragma: no cover
test_mod = importlib.import_module(f"{modules_path}.test_main_003")
return Modules(app=app_mod, test=test_mod)
import pytest
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@dataclass
@pytest.fixture(
name="modules_path",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
app_mod = sys.modules[app_mod_path]
importlib.reload(app_mod)
else:
- app_mod = importlib.import_module(app_mod_path)
+ app_mod = importlib.import_module(app_mod_path) # pragma: no cover
test_mod = importlib.import_module(f"{modules_path}.test_main_004")
return Modules(app=app_mod, test=test_mod)
import pytest
from sqlmodel import Session
-from docs_src.tutorial.fastapi.app_testing.tutorial001 import main as app_mod
-from docs_src.tutorial.fastapi.app_testing.tutorial001 import test_main_005 as test_mod
-from docs_src.tutorial.fastapi.app_testing.tutorial001.test_main_005 import (
+from docs_src.tutorial.fastapi.app_testing.tutorial001_py39 import main as app_mod
+from docs_src.tutorial.fastapi.app_testing.tutorial001_py39 import (
+ test_main_005 as test_mod,
+)
+from docs_src.tutorial.fastapi.app_testing.tutorial001_py39.test_main_005 import (
session_fixture,
)
from fastapi.testclient import TestClient
from sqlmodel import Session
-from docs_src.tutorial.fastapi.app_testing.tutorial001 import main as app_mod
-from docs_src.tutorial.fastapi.app_testing.tutorial001 import test_main_006 as test_mod
-from docs_src.tutorial.fastapi.app_testing.tutorial001.test_main_006 import (
+from docs_src.tutorial.fastapi.app_testing.tutorial001_py39 import main as app_mod
+from docs_src.tutorial.fastapi.app_testing.tutorial001_py39 import (
+ test_main_006 as test_mod,
+)
+from docs_src.tutorial.fastapi.app_testing.tutorial001_py39.test_main_006 import (
client_fixture,
session_fixture,
)
import pytest
-from ....conftest import needs_py39, needs_py310
+from ....conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
from sqlmodel import Session, create_engine
from sqlmodel.pool import StaticPool
-from tests.conftest import needs_py39, needs_py310
+from tests.conftest import needs_py310
@pytest.fixture(
name="module",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
projects / thirdparty / fastapi / sqlmodel.git / commitdiff
