created public api

api/app.py

@@ -0,0 +1,37 @@
+# from contextlib import asynccontextmanager
+
+# from fastapi import FastAPI
+
+# from api.config import Settings
+# from api.database import create_db_and_tables
+# from api.public import api as public_api
+# # from api.utils.logger import logger_config
+# # from api.utils.mock_data_generator import create_heroes_and_teams
+
+# logger = logger_config(__name__)
+
+
+# @asynccontextmanager
+# async def lifespan(app: FastAPI):
+#     create_db_and_tables()
+#     create_heroes_and_teams()
+
+#     logger.info("startup: triggered")
+
+#     yield
+
+#     logger.info("shutdown: triggered")
+
+
+# def create_app(settings: Settings):
+#     app = FastAPI(
+#         title=settings.PROJECT_NAME,
+#         version=settings.VERSION,
+#         docs_url="/",
+#         description=settings.DESCRIPTION,
+#         lifespan=lifespan,
+#     )
+
+#     app.include_router(public_api)
+
+#     return app

api/auth/__init__.py

@@ -0,0 +1,10 @@
+from fastapi_auth0 import Auth0, Auth0User
+from dotenv import load_dotenv
+from os import getenv
+
+load_dotenv
+
+domain = getenv("AUTH0_DOMAIN")
+api_audience = "API_AUDIENCE"
+
+authenticate = Auth0(domain='your-tenant.auth0.com', api_audience='your-api-identifier', scopes={'read:blabla': ''})

api/config.py

@@ -0,0 +1,30 @@
+import os
+import secrets
+from typing import Literal
+
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = f"SQLModel API - {os.getenv('ENV', 'development').capitalize()}"
+    DESCRIPTION: str = "A FastAPI + SQLModel production-ready API"
+    ENV: Literal["development", "staging", "production"] = "development"
+    VERSION: str = "0.1"
+    SECRET_KEY: str = secrets.token_urlsafe(32)
+    DATABASE_URI: str = "sqlite:////Users/anth/dev/fastapi-sqlmodel/database.db"
+    API_USERNAME: str = "svc_test"
+    API_PASSWORD: str = "superstrongpassword"
+
+    class Config:
+        case_sensitive = True
+
+
+settings = Settings()
+
+
+class TestSettings(Settings):
+    class Config:
+        case_sensitive = True
+
+
+test_settings = TestSettings()

api/database.py

@@ -0,0 +1,15 @@
+from sqlmodel import Session, SQLModel, create_engine
+
+from api.config import settings
+
+connect_args = {"check_same_thread": False}
+engine = create_engine(settings.DATABASE_URI, echo=True, connect_args=connect_args)
+
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+
+
+def get_db():
+    with Session(engine) as session:
+        yield session

api/public/people/crud.py

@@ -0,0 +1,35 @@
+from typing import Optional, List
+from sqlmodel import Session, select
+from api.public.people.models import Person, PersonCreate, PersonUpdate
+
+
+def create_person(db: Session, person: PersonCreate) -> Person:
+    db_person = Person(**person.model_dump())
+    db.add(db_person)
+    db.commit()
+    db.refresh(db_person)
+    return db_person
+
+
+def get_person(db: Session, person_id: int) -> Optional[Person]:
+    return db.exec(Person).filter(Person.id == person_id).first()
+
+
+def get_people(db: Session, skip: int = 0, limit: int = 10) -> List[Person]:
+    return db.exec(Person).offset(skip).limit(limit).all()
+
+
+def update_person(db: Session, person: Person, updated_person: PersonUpdate) -> Person:
+    for key, value in updated_person.model_dump(exclude_unset=True).items():
+        setattr(person, key, value)
+    db.add(person)
+    db.commit()
+    db.refresh(person)
+    return person
+
+
+def delete_person(db: Session, person_id: int) -> Person:
+    person = db.exec(Person).filter(Person.id == person_id).first()
+    db.delete(person)
+    db.commit()
+    return person

api/public/people/models.py

@@ -0,0 +1,29 @@
+from typing import Optional, List, TYPE_CHECKING
+from sqlmodel import SQLModel, Field, Relationship
+
+if TYPE_CHECKING:
+    from api.public.towns.models import Town
+
+class PersonBase(SQLModel):
+    name: str = Field(index=True)
+    age: Optional[int] = Field(default=None, index=True)
+    town_id: Optional[int] = Field(default=None, foreign_key="town.id")
+
+
+
+class Person(PersonBase, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+
+    towns = List["Town"] = Relationship(back_populates="people")
+
+class PersonCreate(PersonBase):
+    pass
+
+class PersonRead(PersonBase):
+    id: int
+
+class PersonUpdate(SQLModel):
+    name: Optional[str] = None
+    age: Optional[int] = None
+    town_id: Optional[int] = None
+

api/public/towns/crud.py

@@ -0,0 +1,72 @@
+from typing import Optional, List
+from sqlmodel import Session
+from api.public.towns.models import Town, TownCreate, TownUpdate
+
+
+def create_town(db: Session, town: TownCreate) -> Town:
+    """
+    The create_town function creates a new town in the database.
+
+    :param db: Session: Pass a database session to the function
+    :param town: TownCreate: Create a new town object
+    :return: The town object that was created
+    """
+    db_town = Town(**town.model_dump())
+    db.add(db_town)
+    db.commit()
+    db.refresh(db_town)
+    return db_town
+
+
+def get_town(db: Session, town_id: int) -> Optional[Town]:
+    """
+    The get_town function returns a town object from the database.
+
+    :param db: Session: Pass the database session into the function
+    :param town_id: int: Filter the town by id
+    :return: An object of the town class
+    """
+    return db.exec(Town).filter(Town.id == town_id).first()
+
+
+def get_towns(db: Session, skip: int = 0, limit: int = 10) -> List[Town]:
+    """
+    The get_towns function returns a list of towns from the database.
+
+    :param db: Session: Pass the database session to the function
+    :param skip: int: Skip a number of rows in the database
+    :param limit: int: Limit the number of results returned
+    :return: A list of town objects
+    """
+    return db.exec(Town).offset(skip).limit(limit).all()
+
+
+def update_town(db: Session, town: Town, updated_town: TownUpdate) -> Town:
+    """
+    The update_town function updates a town in the database.
+
+    :param db: Session: Pass in the database session
+    :param town: Town: Get the town that is being updated
+    :param updated_town: TownUpdate: Pass in the updated values of the town
+    :return: The updated town
+    """
+    for key, value in updated_town.model_dump(exclude_unset=True).items():
+        setattr(town, key, value)
+    db.add(town)
+    db.commit()
+    db.refresh(town)
+    return town
+
+
+def delete_town(db: Session, town_id: int) -> Town:
+    """
+    The delete_town function deletes a town from the database.
+
+    :param db: Session: Connect to the database
+    :param town_id: int: Specify which town to delete
+    :return: The deleted town
+    """
+    town = db.exec(Town).filter(Town.id == town_id).first()
+    db.delete(town)
+    db.commit()
+    return town

api/public/towns/models.py

@@ -0,0 +1,34 @@
+from typing import Optional, List, TYPE_CHECKING
+from sqlmodel import SQLModel, Field, Relationship
+
+if TYPE_CHECKING:
+    from api.public.people.models import Person
+
+
+class TownBase(SQLModel):
+    """Base Town class"""
+    name: str = Field(index=True)
+    population: int = Field(default=None, index=True)
+    country: Optional[str] = Field(default=None)
+
+class Town(TownBase, table=True):
+    """Town Table"""
+    id: Optional[int] = Field(default=None, primary_key=True)
+
+    people: List["Person"] = Relationship(back_populates="towns")
+
+
+class TownCreate(TownBase):
+    """Create a town"""
+    pass
+
+class TownRead(TownBase):
+    """Read details of a town."""
+    id: int
+
+class TownUpdate(SQLModel):
+    """Update details of a town."""
+    name: Optional[str] = None
+    population: Optional[int] = None
+    country: Optional[str] = None
+

api/public/towns/routes.py

@@ -0,0 +1,59 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from typing import List
+
+from api.database import get_db
+from api.public.towns.crud import create_town, get_town, get_towns, update_town, delete_town
+from api.public.towns.models import TownCreate, TownRead, TownUpdate
+
+router = APIRouter(prefix="/towns", tags=["Towns"])
+
+
+@router.post("/", response_model=TownRead)
+def create_new_town(town: TownCreate, db: Session = Depends(get_db)):
+    return create_town(db, town)
+
+
+@router.get("/{town_id}", response_model=TownRead)
+def get_single_town(town_id: int, db: Session = Depends(get_db)):
+    town = get_town(db, town_id)
+    if town is None:
+        raise HTTPException(status_code=404, detail="Town not found")
+    return town
+
+
+@router.get("/", response_model=List[TownRead])
+def get_all_towns(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
+    return get_towns(db, skip=skip, limit=limit)
+
+
+@router.put("/{town_id}", response_model=TownRead)
+def update_existing_town(town_id: int, town: TownUpdate, db: Session = Depends(get_db)):
+    """
+    The update_existing_town function updates an existing town in the database.
+
+    :param town_id: int: Identify the town to be updated
+    :param town: TownUpdate: Pass in the new data for the town
+    :param db: Session: Pass the database session to the function
+    :return: The updated town object
+    """
+    existing_town = get_town(db, town_id)
+    if existing_town is None:
+        raise HTTPException(status_code=404, detail="Town not found")
+    return update_town(db, existing_town, town)
+
+
+@router.delete("/{town_id}", response_model=TownRead)
+def delete_existing_town(town_id: int, db: Session = Depends(get_db)):
+    """
+    The delete_existing_town function deletes a town from the database.
+        It takes in an integer, town_id, and returns a boolean value indicating whether or not the deletion was successful.
+
+    :param town_id: int: Specify the id of the town to be deleted
+    :param db: Session: Pass in the database session
+    :return: The result of delete_town
+    """
+    town = get_town(db, town_id)
+    if town is None:
+        raise HTTPException(status_code=404, detail="Town not found")
+    return delete_town(db, town_id)

api/public/towns/urls.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+from api.public.towns.routes import towns
+
+router = APIRouter()
+
+router.include_router(towns.router, prefix="/towns")