pydantic-version:
- pydantic-v2
include:
- - os: macos-latest
- python-version: "3.8"
- pydantic-version: pydantic-v1
- os: windows-latest
python-version: "3.9"
pydantic-version: pydantic-v2
* If the PR is for a feature, it should have docs.
* Unless it's a feature we want to discourage, like support for a corner case that we don't want users to use.
* The docs should include a source example file, not write Python directly in Markdown.
-* If the source example(s) file can have different syntax for Python 3.8, 3.9, 3.10, there should be different versions of the file, and they should be shown in tabs in the docs.
+* If the source example(s) file can have different syntax for different Python versions, there should be different versions of the file, and they should be shown in tabs in the docs.
* There should be tests testing the source example.
* Before the PR is applied, the new tests should fail.
* After applying the PR, the new tests should pass.
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```Python
-{!./docs_src/tutorial/automatic_id_none_refresh/tutorial002.py!}
+{!./docs_src/tutorial/automatic_id_none_refresh/tutorial002_py39.py!}
```
{!./docs_src/tutorial/automatic_id_none_refresh/annotations/en/tutorial002.md!}
You could put all the database Models in a single Python module (a single Python file), for example `models.py`:
-```Python
-{!./docs_src/tutorial/code_structure/tutorial001/models.py!}
-```
+{* ./docs_src/tutorial/code_structure/tutorial001_py310/models.py *}
This way, you wouldn't have to deal with circular imports for other models.
Then you could put the code creating the **engine** and the function to create all the tables (if you are not using migrations) in another file `database.py`:
-```Python
-{!./docs_src/tutorial/code_structure/tutorial001/database.py!}
-```
+{* ./docs_src/tutorial/code_structure/tutorial001_py310/database.py *}
This file would also be imported by your application code, to use the shared **engine** and to get and call the function `create_db_and_tables()`.
Finally, you could put the code to create the **app** in another file `app.py`:
-```Python hl_lines="3-4"
-{!./docs_src/tutorial/code_structure/tutorial001/app.py!}
-```
+{* ./docs_src/tutorial/code_structure/tutorial001_py310/app.py hl[3:4] *}
Here we import the models, the engine, and the function to create all the tables and then we can use them all internally.
Using that trick of `TYPE_CHECKING` we can "import" the `Team` in `hero_model.py`:
-```Python hl_lines="1 5-6 16"
-{!./docs_src/tutorial/code_structure/tutorial002/hero_model.py!}
-```
+{* ./docs_src/tutorial/code_structure/tutorial002_py310/hero_model.py hl[1,5:6,16] *}
Have in mind that now we *have* to put the annotation of `Team` as a string: `"Team"`, so that Python doesn't have errors at runtime.
We use the same trick in the `team_model.py` file:
-```Python hl_lines="1 5-6 14"
-{!./docs_src/tutorial/code_structure/tutorial002/team_model.py!}
-```
+{* ./docs_src/tutorial/code_structure/tutorial002_py310/team_model.py hl[1,5:6,14] *}
Now we get editor support, autocompletion, inline errors, and **SQLModel** keeps working. 🎉
Now, just for completeness, the `app.py` file would import the models from both modules:
-```Python hl_lines="4-5 10 12-14"
-{!./docs_src/tutorial/code_structure/tutorial002/app.py!}
-```
+{* ./docs_src/tutorial/code_structure/tutorial002_py310/app.py hl[4:5,10,12:14] *}
And of course, all the tricks with `TYPE_CHECKING` and type annotations in strings are **only needed in the files with circular imports**.
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{.python .annotate}
-{!./docs_src/tutorial/create_db_and_table/tutorial003.py!}
+{!./docs_src/tutorial/create_db_and_table/tutorial003_py39.py!}
```
{!./docs_src/tutorial/create_db_and_table/annotations/en/tutorial003.md!}
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{ .python .annotate hl_lines="72-90" }
-{!./docs_src/tutorial/delete/tutorial002.py!}
+{!./docs_src/tutorial/delete/tutorial002_py39.py!}
```
{!./docs_src/tutorial/delete/annotations/en/tutorial002.md!}
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
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{.python .annotate hl_lines="22" }
-{!./docs_src/tutorial/create_db_and_table/tutorial003.py[ln:1-20]!}
+{!./docs_src/tutorial/create_db_and_table/tutorial003_py39.py[ln:1-20]!}
# More code here later 👈
-{!./docs_src/tutorial/create_db_and_table/tutorial003.py[ln:23-24]!}
+{!./docs_src/tutorial/create_db_and_table/tutorial003_py39.py[ln:23-24]!}
```
{!./docs_src/tutorial/create_db_and_table/annotations/en/tutorial003.md!}
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{.python .annotate }
-{!./docs_src/tutorial/insert/tutorial003.py!}
+{!./docs_src/tutorial/insert/tutorial003_py39.py!}
```
{!./docs_src/tutorial/insert/annotations/en/tutorial003.md!}
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{ .python .annotate }
-{!./docs_src/tutorial/select/tutorial002.py!}
+{!./docs_src/tutorial/select/tutorial002_py39.py!}
```
{!./docs_src/tutorial/select/annotations/en/tutorial002.md!}
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{ .python .annotate hl_lines="44-55" }
-{!./docs_src/tutorial/update/tutorial002.py!}
+{!./docs_src/tutorial/update/tutorial002_py39.py!}
```
{!./docs_src/tutorial/update/annotations/en/tutorial002.md!}
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```{ .python .annotate hl_lines="15-17 19-21 23" }
# Code above omitted 👆
-{!./docs_src/tutorial/update/tutorial004.py[ln:44-70]!}
+{!./docs_src/tutorial/update/tutorial004_py39.py[ln:44-70]!}
# Code below omitted 👇
```
////
-//// tab | Python 3.8+
+//// tab | Python 3.9+
```Python
-{!./docs_src/tutorial/update/tutorial004.py!}
+{!./docs_src/tutorial/update/tutorial004_py39.py!}
```
////
+++ /dev/null
-from sqlmodel import Session
-
-from .database import create_db_and_tables, engine
-from .models import Hero, Team
-
-
-def create_heroes():
- with Session(engine) as session:
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- session.add(hero_deadpond)
- session.commit()
-
- session.refresh(hero_deadpond)
-
- print("Created hero:", hero_deadpond)
- print("Hero's team:", hero_deadpond.team)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from sqlmodel import SQLModel, create_engine
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, SQLModel
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
+++ /dev/null
-from sqlmodel import Session
-
-from .database import create_db_and_tables, engine
-from .hero_model import Hero
-from .team_model import Team
-
-
-def create_heroes():
- with Session(engine) as session:
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- session.add(hero_deadpond)
- session.commit()
-
- session.refresh(hero_deadpond)
-
- print("Created hero:", hero_deadpond)
- print("Hero's team:", hero_deadpond.team)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from sqlmodel import SQLModel, create_engine
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
+++ /dev/null
-from typing import TYPE_CHECKING, Optional
-
-from sqlmodel import Field, Relationship, SQLModel
-
-if TYPE_CHECKING:
- from .team_model import Team
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional["Team"] = Relationship(back_populates="heroes")
+++ /dev/null
-from typing import TYPE_CHECKING, List, Optional
-
-from sqlmodel import Field, Relationship, SQLModel
-
-if TYPE_CHECKING:
- from .hero_model import Hero
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
4. Create the `id` field:
- It could be `None` until the database assigns a value to it, so we annotate it with `Optional`.
+ It could be `None` until the database assigns a value to it, so we annotate it with `Optional` (`int | None` in Python 3.10+).
It is a **primary key**, so we use `Field()` and the argument `primary_key=True`.
In the database, the default value will be `NULL`, the SQL equivalent of `None`.
- As this field could be `None` (and `NULL` in the database), we annotate it with `Optional`.
+ As this field could be `None` (and `NULL` in the database), we annotate it with `Optional` (`int | None` in Python 3.10+).
8. Write the name of the database file.
9. Use the name of the database file to create the database URL.
+++ /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
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
-
-
-class HeroTeamLink(SQLModel, table=True):
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", primary_key=True
- )
- hero_id: Optional[int] = Field(
- default=None, foreign_key="hero.id", primary_key=True
- )
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="teams", link_model=HeroTeamLink)
-
-
-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)
-
- teams: List[Team] = Relationship(back_populates="heroes", link_model=HeroTeamLink)
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond",
- secret_name="Dive Wilson",
- teams=[team_z_force, team_preventers],
- )
- hero_rusty_man = Hero(
- name="Rusty-Man",
- secret_name="Tommy Sharp",
- age=48,
- teams=[team_preventers],
- )
- hero_spider_boy = Hero(
- name="Spider-Boy", secret_name="Pedro Parqueador", teams=[team_preventers]
- )
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Deadpond:", hero_deadpond)
- print("Deadpond teams:", hero_deadpond.teams)
- print("Rusty-Man:", hero_rusty_man)
- print("Rusty-Man Teams:", hero_rusty_man.teams)
- print("Spider-Boy:", hero_spider_boy)
- print("Spider-Boy Teams:", hero_spider_boy.teams)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class HeroTeamLink(SQLModel, table=True):
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", primary_key=True
- )
- hero_id: Optional[int] = Field(
- default=None, foreign_key="hero.id", primary_key=True
- )
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="teams", link_model=HeroTeamLink)
-
-
-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)
-
- teams: List[Team] = Relationship(back_populates="heroes", link_model=HeroTeamLink)
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond",
- secret_name="Dive Wilson",
- teams=[team_z_force, team_preventers],
- )
- hero_rusty_man = Hero(
- name="Rusty-Man",
- secret_name="Tommy Sharp",
- age=48,
- teams=[team_preventers],
- )
- hero_spider_boy = Hero(
- name="Spider-Boy", secret_name="Pedro Parqueador", teams=[team_preventers]
- )
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Deadpond:", hero_deadpond)
- print("Deadpond teams:", hero_deadpond.teams)
- print("Rusty-Man:", hero_rusty_man)
- print("Rusty-Man Teams:", hero_rusty_man.teams)
- print("Spider-Boy:", hero_spider_boy)
- print("Spider-Boy Teams:", hero_spider_boy.teams)
-
-
-def update_heroes():
- with Session(engine) as session:
- hero_spider_boy = session.exec(
- select(Hero).where(Hero.name == "Spider-Boy")
- ).one()
- team_z_force = session.exec(select(Team).where(Team.name == "Z-Force")).one()
-
- team_z_force.heroes.append(hero_spider_boy)
- session.add(team_z_force)
- session.commit()
-
- print("Updated Spider-Boy's Teams:", hero_spider_boy.teams)
- print("Z-Force heroes:", team_z_force.heroes)
-
- hero_spider_boy.teams.remove(team_z_force)
- session.add(team_z_force)
- session.commit()
-
- print("Reverted Z-Force's heroes:", team_z_force.heroes)
- print("Reverted Spider-Boy's teams:", hero_spider_boy.teams)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- update_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class HeroTeamLink(SQLModel, table=True):
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", primary_key=True
- )
- hero_id: Optional[int] = Field(
- default=None, foreign_key="hero.id", primary_key=True
- )
- is_training: bool = False
-
- team: "Team" = Relationship(back_populates="hero_links")
- hero: "Hero" = Relationship(back_populates="team_links")
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- hero_links: List[HeroTeamLink] = Relationship(back_populates="team")
-
-
-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)
-
- team_links: List[HeroTeamLink] = Relationship(back_populates="hero")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond",
- secret_name="Dive Wilson",
- )
- hero_rusty_man = Hero(
- name="Rusty-Man",
- secret_name="Tommy Sharp",
- age=48,
- )
- hero_spider_boy = Hero(
- name="Spider-Boy",
- secret_name="Pedro Parqueador",
- )
- deadpond_team_z_link = HeroTeamLink(team=team_z_force, hero=hero_deadpond)
- deadpond_preventers_link = HeroTeamLink(
- team=team_preventers, hero=hero_deadpond, is_training=True
- )
- spider_boy_preventers_link = HeroTeamLink(
- team=team_preventers, hero=hero_spider_boy, is_training=True
- )
- rusty_man_preventers_link = HeroTeamLink(
- team=team_preventers, hero=hero_rusty_man
- )
-
- session.add(deadpond_team_z_link)
- session.add(deadpond_preventers_link)
- session.add(spider_boy_preventers_link)
- session.add(rusty_man_preventers_link)
- session.commit()
-
- for link in team_z_force.hero_links:
- print("Z-Force hero:", link.hero, "is training:", link.is_training)
-
- for link in team_preventers.hero_links:
- print("Preventers hero:", link.hero, "is training:", link.is_training)
-
-
-def update_heroes():
- with Session(engine) as session:
- hero_spider_boy = session.exec(
- select(Hero).where(Hero.name == "Spider-Boy")
- ).one()
- team_z_force = session.exec(select(Team).where(Team.name == "Z-Force")).one()
-
- spider_boy_z_force_link = HeroTeamLink(
- team=team_z_force, hero=hero_spider_boy, is_training=True
- )
- team_z_force.hero_links.append(spider_boy_z_force_link)
- session.add(team_z_force)
- session.commit()
-
- print("Updated Spider-Boy's Teams:", hero_spider_boy.team_links)
- print("Z-Force heroes:", team_z_force.hero_links)
-
- for link in hero_spider_boy.team_links:
- if link.team.name == "Preventers":
- link.is_training = False
-
- session.add(hero_spider_boy)
- session.commit()
-
- for link in hero_spider_boy.team_links:
- print("Spider-Boy team:", link.team, "is training:", link.is_training)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- update_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship()
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship()
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
- hero_tarantula = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
- hero_dr_weird = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
- hero_cap = Hero(
- name="Captain North America", secret_name="Esteban Rogelios", age=93
- )
-
- team_preventers.heroes.append(hero_tarantula)
- team_preventers.heroes.append(hero_dr_weird)
- team_preventers.heroes.append(hero_cap)
- session.add(team_preventers)
- session.commit()
- session.refresh(hero_tarantula)
- session.refresh(hero_dr_weird)
- session.refresh(hero_cap)
- print("Preventers new hero:", hero_tarantula)
- print("Preventers new hero:", hero_dr_weird)
- print("Preventers new hero:", hero_cap)
-
-
-def select_heroes():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Preventers")
- result = session.exec(statement)
- team_preventers = result.one()
-
- print("Preventers heroes:", team_preventers.heroes)
-
-
-def update_heroes():
- with Session(engine) as session:
- hero_spider_boy = session.exec(
- select(Hero).where(Hero.name == "Spider-Boy")
- ).one()
-
- preventers_team = session.exec(
- select(Team).where(Team.name == "Preventers")
- ).one()
-
- print("Hero Spider-Boy:", hero_spider_boy)
- print("Preventers Team:", preventers_team)
- print("Preventers Team Heroes:", preventers_team.heroes)
-
- hero_spider_boy.team = None
-
- print("Spider-Boy without team:", hero_spider_boy)
-
- print("Preventers Team Heroes again:", preventers_team.heroes)
-
- session.add(hero_spider_boy)
- session.commit()
- print("After committing")
-
- session.refresh(hero_spider_boy)
- print("Spider-Boy after commit:", hero_spider_boy)
-
- print("Preventers Team Heroes after commit:", preventers_team.heroes)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- select_heroes()
- update_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
- hero_tarantula = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
- hero_dr_weird = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
- hero_cap = Hero(
- name="Captain North America", secret_name="Esteban Rogelios", age=93
- )
-
- team_preventers.heroes.append(hero_tarantula)
- team_preventers.heroes.append(hero_dr_weird)
- team_preventers.heroes.append(hero_cap)
- session.add(team_preventers)
- session.commit()
- session.refresh(hero_tarantula)
- session.refresh(hero_dr_weird)
- session.refresh(hero_cap)
- print("Preventers new hero:", hero_tarantula)
- print("Preventers new hero:", hero_dr_weird)
- print("Preventers new hero:", hero_cap)
-
-
-def select_heroes():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Preventers")
- result = session.exec(statement)
- team_preventers = result.one()
-
- print("Preventers heroes:", team_preventers.heroes)
-
-
-def update_heroes():
- with Session(engine) as session:
- hero_spider_boy = session.exec(
- select(Hero).where(Hero.name == "Spider-Boy")
- ).one()
-
- preventers_team = session.exec(
- select(Team).where(Team.name == "Preventers")
- ).one()
-
- print("Hero Spider-Boy:", hero_spider_boy)
- print("Preventers Team:", preventers_team)
- print("Preventers Team Heroes:", preventers_team.heroes)
-
- hero_spider_boy.team = None
-
- print("Spider-Boy without team:", hero_spider_boy)
-
- print("Preventers Team Heroes again:", preventers_team.heroes)
-
- session.add(hero_spider_boy)
- session.commit()
- print("After committing")
-
- session.refresh(hero_spider_boy)
- print("Spider-Boy after commit:", hero_spider_boy)
-
- print("Preventers Team Heroes after commit:", preventers_team.heroes)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- select_heroes()
- update_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, SQLModel, create_engine
-
-
-class Weapon(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
-
- hero: "Hero" = Relationship(back_populates="weapon")
-
-
-class Power(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
-
- hero_id: int = Field(foreign_key="hero.id")
- hero: "Hero" = Relationship(back_populates="powers")
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
-
- weapon_id: Optional[int] = Field(default=None, foreign_key="weapon.id")
- weapon: Optional[Weapon] = Relationship(back_populates="hero")
-
- powers: List[Power] = Relationship(back_populates="hero")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def main():
- create_db_and_tables()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team", cascade_delete=True)
-
-
-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)
-
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", ondelete="CASCADE"
- )
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
-
-def delete_team():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Wakaland")
- team = session.exec(statement).one()
- session.delete(team)
- session.commit()
- print("Deleted team:", team)
-
-
-def select_deleted_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Black Lion")
- result = session.exec(statement)
- hero = result.first()
- print("Black Lion not found:", hero)
-
- statement = select(Hero).where(Hero.name == "Princess Sure-E")
- result = session.exec(statement)
- hero = result.first()
- print("Princess Sure-E not found:", hero)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- delete_team()
- select_deleted_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", ondelete="SET NULL"
- )
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
-
-def delete_team():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Wakaland")
- team = session.exec(statement).one()
- session.delete(team)
- session.commit()
- print("Deleted team:", team)
-
-
-def select_deleted_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Black Lion")
- result = session.exec(statement)
- hero = result.first()
- print("Black Lion has no team:", hero)
-
- statement = select(Hero).where(Hero.name == "Princess Sure-E")
- result = session.exec(statement)
- hero = result.first()
- print("Princess Sure-E has no team:", hero)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- delete_team()
- select_deleted_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team", passive_deletes="all")
-
-
-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)
-
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", ondelete="SET NULL"
- )
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
- with engine.connect() as connection:
- connection.execute(text("PRAGMA foreign_keys=ON")) # for SQLite only
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
-
-def delete_team():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Wakaland")
- team = session.exec(statement).one()
- session.delete(team)
- session.commit()
- print("Deleted team:", team)
-
-
-def select_deleted_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Black Lion")
- result = session.exec(statement)
- hero = result.first()
- print("Black Lion has no team:", hero)
-
- statement = select(Hero).where(Hero.name == "Princess Sure-E")
- result = session.exec(statement)
- hero = result.first()
- print("Princess Sure-E has no team:", hero)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- delete_team()
- select_deleted_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team", passive_deletes="all")
-
-
-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)
-
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", ondelete="RESTRICT"
- )
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
- with engine.connect() as connection:
- connection.execute(text("PRAGMA foreign_keys=ON")) # for SQLite only
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
-
-def delete_team():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Wakaland")
- team = session.exec(statement).one()
- session.delete(team)
- session.commit()
- print("Deleted team:", team)
-
-
-def select_deleted_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Black Lion")
- result = session.exec(statement)
- hero = result.first()
- print("Black Lion has no team:", hero)
-
- statement = select(Hero).where(Hero.name == "Princess Sure-E")
- result = session.exec(statement)
- hero = result.first()
- print("Princess Sure-E has no team:", hero)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- delete_team()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select, text
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team", passive_deletes="all")
-
-
-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)
-
- team_id: Optional[int] = Field(
- default=None, foreign_key="team.id", ondelete="RESTRICT"
- )
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
- with engine.connect() as connection:
- connection.execute(text("PRAGMA foreign_keys=ON")) # for SQLite only
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
-
-def remove_team_heroes():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Wakaland")
- team = session.exec(statement).one()
- team.heroes.clear()
- session.add(team)
- session.commit()
- session.refresh(team)
- print("Team with removed heroes:", team)
-
-
-def delete_team():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Wakaland")
- team = session.exec(statement).one()
- session.delete(team)
- session.commit()
- print("Deleted team:", team)
-
-
-def select_deleted_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Black Lion")
- result = session.exec(statement)
- hero = result.first()
- print("Black Lion has no team:", hero)
-
- statement = select(Hero).where(Hero.name == "Princess Sure-E")
- result = session.exec(statement)
- hero = result.first()
- print("Princess Sure-E has no team:", hero)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- remove_team_heroes()
- delete_team()
- select_deleted_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
- hero_tarantula = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
- hero_dr_weird = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
- hero_cap = Hero(
- name="Captain North America", secret_name="Esteban Rogelios", age=93
- )
-
- team_preventers.heroes.append(hero_tarantula)
- team_preventers.heroes.append(hero_dr_weird)
- team_preventers.heroes.append(hero_cap)
- session.add(team_preventers)
- session.commit()
- session.refresh(hero_tarantula)
- session.refresh(hero_dr_weird)
- session.refresh(hero_cap)
- print("Preventers new hero:", hero_tarantula)
- print("Preventers new hero:", hero_dr_weird)
- print("Preventers new hero:", hero_cap)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
-
-
-def main():
- create_db_and_tables()
- create_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
- hero_tarantula = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
- hero_dr_weird = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
- hero_cap = Hero(
- name="Captain North America", secret_name="Esteban Rogelios", age=93
- )
-
- team_preventers.heroes.append(hero_tarantula)
- team_preventers.heroes.append(hero_dr_weird)
- team_preventers.heroes.append(hero_cap)
- session.add(team_preventers)
- session.commit()
- session.refresh(hero_tarantula)
- session.refresh(hero_dr_weird)
- session.refresh(hero_cap)
- print("Preventers new hero:", hero_tarantula)
- print("Preventers new hero:", hero_dr_weird)
- print("Preventers new hero:", hero_cap)
-
-
-def select_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Spider-Boy")
- result = session.exec(statement)
- hero_spider_boy = result.one()
-
- statement = select(Team).where(Team.id == hero_spider_boy.team_id)
- result = session.exec(statement)
- team = result.first()
- print("Spider-Boy's team:", team)
-
- print("Spider-Boy's team again:", hero_spider_boy.team)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- select_heroes()
-
-
-if __name__ == "__main__":
- main()
+++ /dev/null
-from typing import List, Optional
-
-from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
-
-
-class Team(SQLModel, table=True):
- id: Optional[int] = Field(default=None, primary_key=True)
- name: str = Field(index=True)
- headquarters: str
-
- heroes: List["Hero"] = Relationship(back_populates="team")
-
-
-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)
-
- team_id: Optional[int] = Field(default=None, foreign_key="team.id")
- team: Optional[Team] = Relationship(back_populates="heroes")
-
-
-sqlite_file_name = "database.db"
-sqlite_url = f"sqlite:///{sqlite_file_name}"
-
-engine = create_engine(sqlite_url, echo=True)
-
-
-def create_db_and_tables():
- SQLModel.metadata.create_all(engine)
-
-
-def create_heroes():
- with Session(engine) as session:
- team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
- team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
-
- hero_deadpond = Hero(
- name="Deadpond", secret_name="Dive Wilson", team=team_z_force
- )
- hero_rusty_man = Hero(
- name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
- )
- hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
- session.add(hero_deadpond)
- session.add(hero_rusty_man)
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_deadpond)
- session.refresh(hero_rusty_man)
- session.refresh(hero_spider_boy)
-
- print("Created hero:", hero_deadpond)
- print("Created hero:", hero_rusty_man)
- print("Created hero:", hero_spider_boy)
-
- hero_spider_boy.team = team_preventers
- session.add(hero_spider_boy)
- session.commit()
- session.refresh(hero_spider_boy)
- print("Updated hero:", hero_spider_boy)
-
- hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
- hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
- team_wakaland = Team(
- name="Wakaland",
- headquarters="Wakaland Capital City",
- heroes=[hero_black_lion, hero_sure_e],
- )
- session.add(team_wakaland)
- session.commit()
- session.refresh(team_wakaland)
- print("Team Wakaland:", team_wakaland)
-
- hero_tarantula = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
- hero_dr_weird = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
- hero_cap = Hero(
- name="Captain North America", secret_name="Esteban Rogelios", age=93
- )
-
- team_preventers.heroes.append(hero_tarantula)
- team_preventers.heroes.append(hero_dr_weird)
- team_preventers.heroes.append(hero_cap)
- session.add(team_preventers)
- session.commit()
- session.refresh(hero_tarantula)
- session.refresh(hero_dr_weird)
- session.refresh(hero_cap)
- print("Preventers new hero:", hero_tarantula)
- print("Preventers new hero:", hero_dr_weird)
- print("Preventers new hero:", hero_cap)
-
-
-def select_heroes():
- with Session(engine) as session:
- statement = select(Team).where(Team.name == "Preventers")
- result = session.exec(statement)
- team_preventers = result.one()
-
- print("Preventers heroes:", team_preventers.heroes)
-
-
-def update_heroes():
- with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Spider-Boy")
- result = session.exec(statement)
- hero_spider_boy = result.one()
-
- hero_spider_boy.team = None
- session.add(hero_spider_boy)
- session.commit()
-
- session.refresh(hero_spider_boy)
- print("Spider-Boy without team:", hero_spider_boy)
-
-
-def main():
- create_db_and_tables()
- create_heroes()
- select_heroes()
- update_heroes()
-
-
-if __name__ == "__main__":
- main()
-r requirements-docs-tests.txt
pytest >=7.0.1,<9.0.0
coverage[toml] >=6.2,<8.0
-# Remove when support for Python 3.8 is dropped
-mypy ==1.14.1; python_version < "3.9"
-mypy ==1.19.1; python_version >= "3.9"
+mypy ==1.19.1
ruff ==0.14.10
# For FastAPI tests
fastapi >=0.103.2,<0.126.0
httpx ==0.28.1
dirty-equals ==0.9.0
jinja2 ==3.1.6
-# Remove when support for Python 3.8 is dropped
-typing-extensions ==4.13.2; python_version < "3.9"
-typing-extensions ==4.15.0; python_version >= "3.9"
+typing-extensions ==4.15.0
def cov_tmp_path(tmp_path: Path) -> Generator[Path, None, None]:
yield tmp_path
for coverage_path in tmp_path.glob(".coverage*"):
- coverage_destiny_path = top_level_path / coverage_path.name
+ coverage_destiny_path = top_level_path / "coverage" / coverage_path.name
shutil.copy(coverage_path, coverage_destiny_path)
needs_pydanticv2 = pytest.mark.skipif(not IS_PYDANTIC_V2, reason="requires Pydantic v2")
needs_pydanticv1 = pytest.mark.skipif(IS_PYDANTIC_V2, reason="requires Pydantic v1")
-needs_py39 = pytest.mark.skipif(sys.version_info < (3, 9), reason="requires python3.9+")
needs_py310 = pytest.mark.skipif(
sys.version_info < (3, 10), reason="requires python3.10+"
)
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
import sys
from pathlib import Path
-from .conftest import needs_py39
-
root_path = Path(__file__).parent.parent
-@needs_py39
def test_select_gen() -> None:
env = os.environ.copy()
env["CHECK_JINJA"] = "1"
@pytest.fixture(
name="module",
params=[
- "tutorial001",
- "tutorial002",
+ "tutorial001_py39",
+ "tutorial002_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
pytest.param("tutorial002_py310", marks=needs_py310),
],
import pytest
from sqlmodel import create_engine
-from tests.conftest import PrintMock, needs_py39, needs_py310
+from tests.conftest import PrintMock, needs_py310
expected_calls = [
[
@pytest.fixture(
name="modules",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ...conftest import PrintMock, needs_py39, needs_py310
+from ...conftest import PrintMock, needs_py310
expected_calls = [
[
@pytest.fixture(
name="modules",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.mark.parametrize(
"module",
[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial002",
+ "tutorial002_py39",
pytest.param("tutorial002_py310", marks=needs_py310),
],
indirect=True,
@pytest.fixture(
name="module",
params=[
- "tutorial003",
+ "tutorial003_py39",
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial004",
+ "tutorial004_py39",
pytest.param("tutorial004_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial005",
+ "tutorial005_py39",
pytest.param("tutorial005_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.mark.parametrize(
"module_name",
[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial002",
+ "tutorial002_py39",
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="module",
params=[
- "tutorial003",
+ "tutorial003_py39",
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.mark.parametrize(
"module",
[
- "tutorial001",
+ "tutorial001_py39",
pytest.param("tutorial001_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial002",
+ "tutorial002_py39",
pytest.param("tutorial002_py310", marks=needs_py310),
],
indirect=True,
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),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial004",
+ pytest.param("tutorial004_py39"),
pytest.param("tutorial004_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ...conftest import PrintMock, needs_py39, needs_py310
+from ...conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ...conftest import PrintMock, needs_py39, needs_py310
+from ...conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ...conftest import PrintMock, needs_py39, needs_py310
+from ...conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
- pytest.param("tutorial003_py39", marks=needs_py39),
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial004",
+ pytest.param("tutorial004_py39"),
pytest.param("tutorial004_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial005",
+ pytest.param("tutorial005_py39"),
pytest.param("tutorial005_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial006",
+ pytest.param("tutorial006_py39"),
pytest.param("tutorial006_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial007",
+ pytest.param("tutorial007_py39"),
pytest.param("tutorial007_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial008",
+ pytest.param("tutorial008_py39"),
pytest.param("tutorial008_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial009",
+ pytest.param("tutorial009_py39"),
pytest.param("tutorial009_py310", marks=needs_py310),
],
)
from sqlalchemy.exc import SAWarning
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
from sqlalchemy.engine.reflection import Inspector
from sqlmodel import create_engine
-from ....conftest import needs_py39, needs_py310
+from ....conftest import needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
- pytest.param("tutorial003_py39", marks=needs_py39),
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
- pytest.param("tutorial003_py39", marks=needs_py39),
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
from sqlalchemy.exc import IntegrityError
from sqlmodel import Session, create_engine, select
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial004",
- pytest.param("tutorial004_py39", marks=needs_py39),
+ pytest.param("tutorial004_py39"),
pytest.param("tutorial004_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial005",
- pytest.param("tutorial005_py39", marks=needs_py39),
+ pytest.param("tutorial005_py39"),
pytest.param("tutorial005_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
- pytest.param("tutorial001_py39", marks=needs_py39),
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
import pytest
from sqlmodel import create_engine
-from ....conftest import PrintMock, needs_py39, needs_py310
+from ....conftest import PrintMock, needs_py310
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
- pytest.param("tutorial002_py39", marks=needs_py39),
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.mark.parametrize(
"module",
[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial003",
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial004",
+ pytest.param("tutorial004_py39"),
pytest.param("tutorial004_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial003",
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
indirect=True,
@pytest.mark.parametrize(
"module",
[
- "tutorial004",
+ pytest.param("tutorial004_py39"),
pytest.param("tutorial004_py310", marks=needs_py310),
],
indirect=True,
@pytest.fixture(
name="mod",
params=[
- "tutorial001",
+ pytest.param("tutorial001_py39"),
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial002",
+ pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial003",
+ pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial004",
+ pytest.param("tutorial004_py39"),
pytest.param("tutorial004_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial005",
+ pytest.param("tutorial005_py39"),
pytest.param("tutorial005_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial006",
+ pytest.param("tutorial006_py39"),
pytest.param("tutorial006_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial007",
+ pytest.param("tutorial007_py39"),
pytest.param("tutorial007_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial008",
+ pytest.param("tutorial008_py39"),
pytest.param("tutorial008_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial009",
+ pytest.param("tutorial009_py39"),
pytest.param("tutorial009_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial010",
+ pytest.param("tutorial010_py39"),
pytest.param("tutorial010_py310", marks=needs_py310),
],
)
@pytest.fixture(
name="mod",
params=[
- "tutorial011",
+ pytest.param("tutorial011_py39"),
pytest.param("tutorial011_py310", marks=needs_py310),
],
)
projects / thirdparty / fastapi / sqlmodel.git / commitdiff
