Skip to content

integrate lite-bootstrap #35

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 17, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions app/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,11 @@


if __name__ == "__main__":
granian.Granian(
granian.Granian( # type: ignore[attr-defined]
target="app.application:application",
address="0.0.0.0", # noqa: S104
port=settings.app_port,
interface=Interfaces.ASGI,
log_dictconfig={"root": {"level": "INFO"}} if not settings.debug else {},
log_level=settings.log_level,
loop=Loops.uvloop,
).serve()
16 changes: 8 additions & 8 deletions app/api/decks.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ async def list_decks(
decks_service: DecksService = FromDI(ioc.Dependencies.decks_service),
) -> schemas.Decks:
objects = await decks_service.list()
return typing.cast(schemas.Decks, {"items": objects})
return typing.cast("schemas.Decks", {"items": objects})


@ROUTER.get("/decks/{deck_id}/")
Expand All @@ -33,7 +33,7 @@ async def get_deck(
if not instance:
raise fastapi.HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Deck is not found")

return typing.cast(schemas.Deck, instance)
return typing.cast("schemas.Deck", instance)


@ROUTER.put("/decks/{deck_id}/")
Expand All @@ -47,7 +47,7 @@ async def update_deck(
except NotFoundError:
raise fastapi.HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Deck is not found") from None

return typing.cast(schemas.Deck, instance)
return typing.cast("schemas.Deck", instance)


@ROUTER.post("/decks/")
Expand All @@ -56,7 +56,7 @@ async def create_deck(
decks_service: DecksService = FromDI(ioc.Dependencies.decks_service),
) -> schemas.Deck:
instance = await decks_service.create(data.model_dump())
return typing.cast(schemas.Deck, instance)
return typing.cast("schemas.Deck", instance)


@ROUTER.get("/decks/{deck_id}/cards/")
Expand All @@ -65,7 +65,7 @@ async def list_cards(
cards_service: CardsService = FromDI(ioc.Dependencies.cards_service),
) -> schemas.Cards:
objects = await cards_service.list(models.Card.deck_id == deck_id)
return typing.cast(schemas.Cards, {"items": objects})
return typing.cast("schemas.Cards", {"items": objects})


@ROUTER.get("/cards/{card_id}/")
Expand All @@ -76,7 +76,7 @@ async def get_card(
instance = await cards_service.get_one_or_none(models.Card.id == card_id)
if not instance:
raise fastapi.HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Card is not found")
return typing.cast(schemas.Card, instance)
return typing.cast("schemas.Card", instance)


@ROUTER.post("/decks/{deck_id}/cards/")
Expand All @@ -88,7 +88,7 @@ async def create_cards(
objects = await cards_service.create_many(
data=[models.Card(**card.model_dump(), deck_id=deck_id) for card in data],
)
return typing.cast(schemas.Cards, {"items": objects})
return typing.cast("schemas.Cards", {"items": objects})


@ROUTER.put("/decks/{deck_id}/cards/")
Expand All @@ -100,4 +100,4 @@ async def update_cards(
objects = await cards_service.upsert_many(
data=[models.Card(**card.model_dump(exclude={"deck_id"}), deck_id=deck_id) for card in data],
)
return typing.cast(schemas.Cards, {"items": objects})
return typing.cast("schemas.Cards", {"items": objects})
38 changes: 20 additions & 18 deletions app/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,48 +4,50 @@
import fastapi
import modern_di_fastapi
from advanced_alchemy.exceptions import DuplicateKeyError
from fastapi.middleware.cors import CORSMiddleware
from lite_bootstrap import FastAPIBootstrapper, FastAPIConfig

from app import exceptions, ioc
from app import exceptions
from app.api.decks import ROUTER
from app.settings import settings


ALLOWED_ORIGINS = [
"http://localhost:5173",
# YOUR ALLOWED ORIGINS HERE
]


def include_routers(app: fastapi.FastAPI) -> None:
app.include_router(ROUTER, prefix="/api")


class AppBuilder:
def __init__(self) -> None:
self.app: fastapi.FastAPI = fastapi.FastAPI(
title=settings.service_name,
debug=settings.debug,
lifespan=self.lifespan_manager,
)
self.bootstrapper = FastAPIBootstrapper(
bootstrap_config=FastAPIConfig(
application=self.app,
service_name=settings.service_name,
service_version=settings.service_version,
service_environment=settings.service_environment,
service_debug=settings.service_debug,
opentelemetry_endpoint=settings.opentelemetry_endpoint,
sentry_dsn=settings.sentry_dsn,
cors_allowed_origins=settings.cors_allowed_origins,
cors_allowed_methods=settings.cors_allowed_methods,
cors_allowed_headers=settings.cors_allowed_headers,
cors_exposed_headers=settings.cors_exposed_headers,
logging_buffer_capacity=settings.logging_buffer_capacity,
swagger_offline_docs=settings.swagger_offline_docs,
),
)
self.bootstrapper.bootstrap()
self.di_container = modern_di_fastapi.setup_di(self.app)
include_routers(self.app)
self.app.add_exception_handler(
DuplicateKeyError,
exceptions.duplicate_key_error_handler, # type: ignore[arg-type]
)
self.app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

@contextlib.asynccontextmanager
async def lifespan_manager(self, _: fastapi.FastAPI) -> typing.AsyncIterator[dict[str, typing.Any]]:
async with self.di_container:
await ioc.Dependencies.async_resolve_creators(self.di_container)
yield {}


Expand Down
4 changes: 2 additions & 2 deletions app/resources/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ async def create_sa_engine() -> typing.AsyncIterator[sa.AsyncEngine]:
logger.info("Initializing SQLAlchemy engine")
engine = sa.create_async_engine(
url=settings.db_dsn,
echo=settings.debug,
echo_pool=settings.debug,
echo=settings.service_debug,
echo_pool=settings.service_debug,
pool_size=settings.db_pool_size,
pool_pre_ping=settings.db_pool_pre_ping,
max_overflow=settings.db_max_overflow,
Expand Down
19 changes: 18 additions & 1 deletion app/settings.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,14 @@
import pydantic
import pydantic_settings
from granian.log import LogLevels
from sqlalchemy.engine.url import URL


class Settings(pydantic_settings.BaseSettings):
service_name: str = "FastAPI template"
debug: bool = False
service_version: str = "1.0.0"
service_environment: str = "local"
service_debug: bool = False
log_level: LogLevels = LogLevels.info

db_driver: str = "postgresql+asyncpg"
Expand All @@ -22,6 +25,20 @@ class Settings(pydantic_settings.BaseSettings):

app_port: int = 8000

opentelemetry_endpoint: str = ""
sentry_dsn: str = ""
logging_buffer_capacity: int = 0
swagger_offline_docs: bool = True

cors_allowed_origins: list[str] = pydantic.Field(
default_factory=lambda: [
"http://localhost:5173",
]
)
cors_allowed_methods: list[str] = pydantic.Field(default_factory=lambda: [""])
cors_allowed_headers: list[str] = pydantic.Field(default_factory=lambda: [""])
cors_exposed_headers: list[str] = pydantic.Field(default_factory=list)

@property
def db_dsn(self) -> URL:
return URL.create(
Expand Down
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,10 @@ authors = [
license = "MIT License"
dependencies = [
"fastapi>=0.76",
"lite-bootstrap[fastapi-all]",
"advanced-alchemy",
"pydantic-settings",
"granian",
"granian[uvloop]",
"modern-di-fastapi",
# database
"alembic",
Expand Down
Loading