def create_heroes():
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (1)
- hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador") # (2)
- hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48) # (3)
-
- print("Before interacting with the database") # (4)
- print("Hero 1:", hero_1) # (5)
- print("Hero 2:", hero_2) # (6)
- print("Hero 3:", hero_3) # (7)
-
- with Session(engine) as session: # (8)
- session.add(hero_1) # (9)
- session.add(hero_2) # (10)
- session.add(hero_3) # (11)
-
- print("After adding to the session") # (12)
- print("Hero 1:", hero_1) # (13)
- print("Hero 2:", hero_2) # (14)
- print("Hero 3:", hero_3) # (15)
-
- session.commit() # (16)
-
- print("After committing the session") # (17)
- print("Hero 1:", hero_1) # (18)
- print("Hero 2:", hero_2) # (19)
- print("Hero 3:", hero_3) # (20)
-
- print("After committing the session, show IDs") # (21)
- print("Hero 1 ID:", hero_1.id) # (22)
- print("Hero 2 ID:", hero_2.id) # (23)
- print("Hero 3 ID:", hero_3.id) # (24)
-
- print("After committing the session, show names") # (25)
- print("Hero 1 name:", hero_1.name) # (26)
- print("Hero 2 name:", hero_2.name) # (27)
- print("Hero 3 name:", hero_3.name) # (28)
-
- session.refresh(hero_1) # (29)
- session.refresh(hero_2) # (30)
- session.refresh(hero_3) # (31)
-
- print("After refreshing the heroes") # (32)
- print("Hero 1:", hero_1) # (33)
- print("Hero 2:", hero_2) # (34)
- print("Hero 3:", hero_3) # (35)
- # (36)
-
- print("After the session closes") # (37)
- print("Hero 1:", hero_1) # (38)
- print("Hero 2:", hero_2) # (39)
- print("Hero 3:", hero_3) # (40)
+ hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (1)!
+ hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador") # (2)!
+ hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48) # (3)!
+
+ print("Before interacting with the database") # (4)!
+ print("Hero 1:", hero_1) # (5)!
+ print("Hero 2:", hero_2) # (6)!
+ print("Hero 3:", hero_3) # (7)!
+
+ with Session(engine) as session: # (8)!
+ session.add(hero_1) # (9)!
+ session.add(hero_2) # (10)!
+ session.add(hero_3) # (11)!
+
+ print("After adding to the session") # (12)!
+ print("Hero 1:", hero_1) # (13)!
+ print("Hero 2:", hero_2) # (14)!
+ print("Hero 3:", hero_3) # (15)!
+
+ session.commit() # (16)!
+
+ print("After committing the session") # (17)!
+ print("Hero 1:", hero_1) # (18)!
+ print("Hero 2:", hero_2) # (19)!
+ print("Hero 3:", hero_3) # (20)!
+
+ print("After committing the session, show IDs") # (21)!
+ print("Hero 1 ID:", hero_1.id) # (22)!
+ print("Hero 2 ID:", hero_2.id) # (23)!
+ print("Hero 3 ID:", hero_3.id) # (24)!
+
+ print("After committing the session, show names") # (25)!
+ print("Hero 1 name:", hero_1.name) # (26)!
+ print("Hero 2 name:", hero_2.name) # (27)!
+ print("Hero 3 name:", hero_3.name) # (28)!
+
+ session.refresh(hero_1) # (29)!
+ session.refresh(hero_2) # (30)!
+ session.refresh(hero_3) # (31)!
+
+ print("After refreshing the heroes") # (32)!
+ print("Hero 1:", hero_1) # (33)!
+ print("Hero 2:", hero_2) # (34)!
+ print("Hero 3:", hero_3) # (35)!
+ # (36)!
+
+ print("After the session closes") # (37)!
+ print("Hero 1:", hero_1) # (38)!
+ print("Hero 2:", hero_2) # (39)!
+ print("Hero 3:", hero_3) # (40)!
def main():
-from typing import Optional # (1)
+from typing import Optional # (1)!
-from sqlmodel import Field, SQLModel, create_engine # (2)
+from sqlmodel import Field, SQLModel, create_engine # (2)!
-class Hero(SQLModel, table=True): # (3)
- id: Optional[int] = Field(default=None, primary_key=True) # (4)
- name: str # (5)
- secret_name: str # (6)
- age: Optional[int] = None # (7)
+class Hero(SQLModel, table=True): # (3)!
+ id: Optional[int] = Field(default=None, primary_key=True) # (4)!
+ name: str # (5)!
+ secret_name: str # (6)!
+ age: Optional[int] = None # (7)!
-sqlite_file_name = "database.db" # (8)
-sqlite_url = f"sqlite:///{sqlite_file_name}" # (9)
+sqlite_file_name = "database.db" # (8)!
+sqlite_url = f"sqlite:///{sqlite_file_name}" # (9)!
-engine = create_engine(sqlite_url, echo=True) # (10)
+engine = create_engine(sqlite_url, echo=True) # (10)!
-def create_db_and_tables(): # (11)
- SQLModel.metadata.create_all(engine) # (12)
+def create_db_and_tables(): # (11)!
+ SQLModel.metadata.create_all(engine) # (12)!
-if __name__ == "__main__": # (13)
- create_db_and_tables() # (14)
+if __name__ == "__main__": # (13)!
+ create_db_and_tables() # (14)!
def delete_heroes():
with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Spider-Youngster") # (1)
- results = session.exec(statement) # (2)
- hero = results.one() # (3)
- print("Hero: ", hero) # (4)
+ statement = select(Hero).where(Hero.name == "Spider-Youngster") # (1)!
+ results = session.exec(statement) # (2)!
+ hero = results.one() # (3)!
+ print("Hero: ", hero) # (4)!
- session.delete(hero) # (5)
- session.commit() # (6)
+ session.delete(hero) # (5)!
+ session.commit() # (6)!
- print("Deleted hero:", hero) # (7)
+ print("Deleted hero:", hero) # (7)!
- statement = select(Hero).where(Hero.name == "Spider-Youngster") # (8)
- results = session.exec(statement) # (9)
- hero = results.first() # (10)
+ statement = select(Hero).where(Hero.name == "Spider-Youngster") # (8)!
+ results = session.exec(statement) # (9)!
+ hero = results.first() # (10)!
- if hero is None: # (11)
- print("There's no hero named Spider-Youngster") # (12)
- # (13)
+ if hero is None: # (11)!
+ print("There's no hero named Spider-Youngster") # (12)!
+ # (13)!
def main():
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
-from .main import app, get_session # (1)
+from .main import app, get_session # (1)!
def test_create_hero():
app.dependency_overrides[get_session] = get_session_override
- client = TestClient(app) # (2)
+ client = TestClient(app) # (2)!
- response = client.post( # (3)
+ response = client.post( # (3)!
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
app.dependency_overrides.clear()
- data = response.json() # (4)
+ 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)
+ 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)!
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
-from .main import app, get_session # (1)
+from .main import app, get_session # (1)!
def test_create_hero():
with Session(engine) as session:
- def get_session_override(): # (2)
- return session # (3)
+ def get_session_override(): # (2)!
+ return session # (3)!
- app.dependency_overrides[get_session] = get_session_override # (4)
+ 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)
+ app.dependency_overrides.clear() # (5)!
data = response.json()
assert response.status_code == 200
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
-from .main import app, get_session # (1)
+from .main import app, get_session # (1)!
def test_create_hero():
- engine = create_engine( # (2)
+ engine = create_engine( # (2)!
"sqlite:///testing.db", connect_args={"check_same_thread": False}
)
- SQLModel.metadata.create_all(engine) # (3)
+ SQLModel.metadata.create_all(engine) # (3)!
- with Session(engine) as session: # (4)
+ with Session(engine) as session: # (4)!
def get_session_override():
- return session # (5)
+ return session # (5)!
- app.dependency_overrides[get_session] = get_session_override # (4)
+ app.dependency_overrides[get_session] = get_session_override # (4)!
client = TestClient(app)
assert data["secret_name"] == "Dive Wilson"
assert data["age"] is None
assert data["id"] is not None
- # (6)
+ # (6)!
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
-from sqlmodel.pool import StaticPool # (1)
+from sqlmodel.pool import StaticPool # (1)!
from .main import app, get_session
def test_create_hero():
engine = create_engine(
- "sqlite://", # (2)
+ "sqlite://", # (2)!
connect_args={"check_same_thread": False},
- poolclass=StaticPool, # (3)
+ poolclass=StaticPool, # (3)!
)
SQLModel.metadata.create_all(engine)
-import pytest # (1)
+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)
+@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)
+ yield session # (4)!
-def test_create_hero(session: Session): # (5)
+def test_create_hero(session: Session): # (5)!
def get_session_override():
- return session # (6)
+ return session # (6)!
app.dependency_overrides[get_session] = get_session_override
yield session
-@pytest.fixture(name="client") # (1)
-def client_fixture(session: Session): # (2)
- def get_session_override(): # (3)
+@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)
+ app.dependency_overrides[get_session] = get_session_override # (4)!
- client = TestClient(app) # (5)
- yield client # (6)
- app.dependency_overrides.clear() # (7)
+ client = TestClient(app) # (5)!
+ yield client # (6)!
+ app.dependency_overrides.clear() # (7)!
-def test_create_hero(client: TestClient): # (8)
+def test_create_hero(client: TestClient): # (8)!
response = client.post(
"/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
)
SQLModel.metadata.create_all(engine)
-def create_heroes(): # (1)
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (2)
+def create_heroes(): # (1)!
+ hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (2)!
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
- with Session(engine) as session: # (3)
- session.add(hero_1) # (4)
+ with Session(engine) as session: # (3)!
+ session.add(hero_1) # (4)!
session.add(hero_2)
session.add(hero_3)
- session.commit() # (5)
- # (6)
+ session.commit() # (5)!
+ # (6)!
-def main(): # (7)
- create_db_and_tables() # (8)
- create_heroes() # (9)
+def main(): # (7)!
+ create_db_and_tables() # (8)!
+ create_heroes() # (9)!
-if __name__ == "__main__": # (10)
- main() # (11)
+if __name__ == "__main__": # (10)!
+ main() # (11)!
from typing import Optional
-from sqlmodel import Field, Session, SQLModel, create_engine, select # (1)
+from sqlmodel import Field, Session, SQLModel, create_engine, select # (1)!
-class Hero(SQLModel, table=True): # (2)
+class Hero(SQLModel, table=True): # (2)!
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
-engine = create_engine(sqlite_url, echo=True) # (3)
+engine = create_engine(sqlite_url, echo=True) # (3)!
def create_db_and_tables():
- SQLModel.metadata.create_all(engine) # (4)
+ SQLModel.metadata.create_all(engine) # (4)!
def create_heroes():
- hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (5)
+ hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (5)!
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
- with Session(engine) as session: # (6)
+ with Session(engine) as session: # (6)!
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
def select_heroes():
- with Session(engine) as session: # (7)
- statement = select(Hero) # (8)
- results = session.exec(statement) # (9)
- for hero in results: # (10)
- print(hero) # (11)
- # (12)
+ with Session(engine) as session: # (7)!
+ statement = select(Hero) # (8)!
+ results = session.exec(statement) # (9)!
+ for hero in results: # (10)!
+ print(hero) # (11)!
+ # (12)!
def main():
create_db_and_tables()
create_heroes()
- select_heroes() # (13)
+ select_heroes() # (13)!
if __name__ == "__main__":
def update_heroes():
with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)
- results = session.exec(statement) # (2)
- hero = results.one() # (3)
- print("Hero:", hero) # (4)
-
- hero.age = 16 # (5)
- session.add(hero) # (6)
- session.commit() # (7)
- session.refresh(hero) # (8)
- print("Updated hero:", hero) # (9)
+ statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
+ results = session.exec(statement) # (2)!
+ hero = results.one() # (3)!
+ print("Hero:", hero) # (4)!
+
+ hero.age = 16 # (5)!
+ session.add(hero) # (6)!
+ session.commit() # (7)!
+ session.refresh(hero) # (8)!
+ print("Updated hero:", hero) # (9)!
def main():
def update_heroes():
with Session(engine) as session:
- statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)
- results = session.exec(statement) # (2)
- hero_1 = results.one() # (3)
- print("Hero 1:", hero_1) # (4)
-
- statement = select(Hero).where(Hero.name == "Captain North America") # (5)
- results = session.exec(statement) # (6)
- hero_2 = results.one() # (7)
- print("Hero 2:", hero_2) # (8)
-
- hero_1.age = 16 # (9)
- hero_1.name = "Spider-Youngster" # (10)
- session.add(hero_1) # (11)
-
- hero_2.name = "Captain North America Except Canada" # (12)
- hero_2.age = 110 # (13)
- session.add(hero_2) # (14)
-
- session.commit() # (15)
- session.refresh(hero_1) # (16)
- session.refresh(hero_2) # (17)
-
- print("Updated hero 1:", hero_1) # (18)
- print("Updated hero 2:", hero_2) # (19)
- # (20)
+ statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
+ results = session.exec(statement) # (2)!
+ hero_1 = results.one() # (3)!
+ print("Hero 1:", hero_1) # (4)!
+
+ statement = select(Hero).where(Hero.name == "Captain North America") # (5)!
+ results = session.exec(statement) # (6)!
+ hero_2 = results.one() # (7)!
+ print("Hero 2:", hero_2) # (8)!
+
+ hero_1.age = 16 # (9)!
+ hero_1.name = "Spider-Youngster" # (10)!
+ session.add(hero_1) # (11)!
+
+ hero_2.name = "Captain North America Except Canada" # (12)!
+ hero_2.age = 110 # (13)!
+ session.add(hero_2) # (14)!
+
+ session.commit() # (15)!
+ session.refresh(hero_1) # (16)!
+ session.refresh(hero_2) # (17)!
+
+ print("Updated hero 1:", hero_1) # (18)!
+ print("Updated hero 2:", hero_2) # (19)!
+ # (20)!
def main():
projects / thirdparty / fastapi / sqlmodel.git / commitdiff
