# fastapi-clean-architecture-boilerplate
## Steps To Use It
Here are the steps to use it:
### Clone The Boilerplate
```bash
git clone https://github.com/Camb-ai/fastapi-clean-architecture-boilerplate
```
### Install The Necessary Pip Packages
Before installing packages we need to create a virtual environment for that you can run:
```bash
python3 -m venv <name>
```
After that you can activate the environment on
unix environment:
```bash
source <name>/bin/activate
```
windows environment:
```bash
<name>/Scripts/activate
```
Now you can start installing all necessary packages by:
```bash
pip install -r requirements.txt
```
### Configure environment variables
Create `.env` file with following contents:
```dotenv
# core
MICROSERVICE_NAME = <name>
# banana.dev
BANANA_DEV_API_KEY = <api-key>
# model keys
# <MODEL_NAME>_MODEL_KEY = <model-key>
# ...other model keys
# open telemetry collector
AGENT_HOSTNAME = "otel"
AGENT_PORT = "4317"
# launch darkly
LAUNCH_DARKLY_SDK_KEY = <sdk-key>
# dramatiq
DRAMATIQ_BROKER = "redis://redis:6379/"
DRAMATIQ_BACKEND = "redis://redis:6379/"
```
### Running with docker compose
You can run the app with docker compose as follows:
```bash
./run-docker-compose.sh
```
Then the FastAPI app would be then available on http://localhost:8000 and Dramatiq Monitoring Dashboard would be available on http://localhost:7640
FastAPI app -
Dramatiq Monitoring Dashboard -
## Local Docker Development With VSCode
### Run App With Docker Compose
First run the app using above mentioned method.
### Install Dev Containers VSCode Extension
### Start VSCode Attached With `fastapi` Container
To do that press `Ctrl + Shift + P` and search dev containers
and click on `Dev Containers: Attach to Running Container...`.
After that select the `fastapi` container
Now a new VSCode window would start with that container attached to it.
After that open the folder fastapi
Now click on remote explorer and then right click on fastapi container and select `show container logs`
After that our VSCode is setup and we can do code editing and debugging in VSCode that would sync with local folder also.
## Creating An Endpoint
There are 2 steps for creating an endpoint:
### Create An Use Case
Use case code can be written inside `src/use_cases/<use-case-name>.py`
There are 2 types of use cases:
#### Synchronous Use Case
```python
from src.use_cases.main import BaseSyncUseCase
from pydantic import BaseModel
class EndPointRequest(BaseModel):
field: fieldtype
# ... more fields
class EndPointResponse(BaseModel):
field: fieldtype
# ... more fields
class EndPointUseCase(BaseSyncUseCase[EndPointRequest, EndPointResponse]):
def validation(self, request_payload: EndPointRequest) -> Exception | None:
# validation logic -> return an exception if validation error occured otherwise return nothing/None.
def handle(self, request_payload: EndPointRequest) -> EndPointResponse:
# use case handling code
# return and object resembling EndPointResponse
end_point_use_case = EndPointUseCase()
```
#### Asynchronous Use Case
```python
from src.use_cases.main import BaseAsyncUseCase
from pydantic import BaseModel
class EndPointRequest(BaseModel):
field: fieldtype
# ... more fields
class EndPointResponse(BaseModel):
field: fieldtype
# ... more fields
class EndPointUseCase(BaseAsyncUseCase[EndPointRequest, EndPointResponse]):
def validation(self, request_payload: EndPointRequest) -> Exception | None:
# validation logic -> return an exception if validation error occured otherwise return nothing/None.
def handle(self, request_payload: EndPointRequest) -> EndPointResponse:
# use case handling code -> this code would run inside task queue
# return and object resembling EndPointResponse
end_point_use_case = EndPointUseCase()
```
### Write Endpoint Code
You can write endpoint code in standard fastapi way:
#### For Synchronous Use Case
```python
@<router-name>.<http-method>(<path>, tags=[tag1, tag2, ....])
async def endpoint(req_body: EndPointRequest):
end_point_use_case.execute(req_body)
return #response
# Asynchronous Use Case -
# Creating task in task queueEndPointUseCase.create(req_body)
#
```
#### For Asynchronous Use Case
##### For Creating Task In Task Queue
```python
@<router-name>.<http-method>(<path>, tags=[tag1, tag2, ....])
async def endpoint(req_body: EndPointRequest):
task = end_point_use_case.create(req_body)
return JSONResponse({"task_id": task.message_id})
```
##### For Getting Task Result
```python
@<router-name>.<http-method>("<path>/{task_id}", tags=[tag1, tag2, ....])
async def endpoint(task_id: str):
task_result= end_point_use_case.get_result(task_id)
return task_result
```
## Creating A Router
There are 2 steps for adding a router in app:
### Writing Router Code
Router code can be written inside `src/external_interfaces/web_server/routers/<router-name>.py` like this:
```python
from fastapi import APIRouter
from src.use_cases.end_point_use_case import end_point_use_case, CreateUserIn
router = APIRouter()
@<router-name>.<http-method>("<path>/{task_id}", tags=[tag1, tag2, ....])
async def endpoint(task_id: str):
task_result= end_point_use_case.get_result(task_id)
return task_result
```
### Adding Router To App
Router can be added to app inside `src/external_interfaces/web_server/main.py` like this:
```python
app.include_router(
<router-name>.router,
prefix="<path>",
tags=[tag1, tag2, ....]
)
```
## Writing Function For Executing `banana.dev` Model
Function for executing banana.dev model can be written in `src/interface_adapters/ml_models/<model-name>.py` like this:
```python
from pydantic import BaseModel
from src.external_interfaces.ml_models import run_ml_model
class ModelOutput(BaseModel):
field: fieldtype
# ... more fields
def model() -> ModelOutput:
model_key = os.getenv("<MODEL_NAME>_MODEL_KEY")
modelpayload = {}
out = run_ml_model(model_key, model_payload)
return ModelOutput.parse_obj(out)
```
Then you can execute `model` in any use case you want. You should always execute it in `AsyncUseCase`.
## Using Feature Flags
You can use `check_flag_value` function as follows:
```python
from src.external_interfaces.monitoring.feature_flags import check_flag_value
user = {
field: value
# other fields related to user
}
default_value = value # can be of any type. It is returned when flag does not exists.
flag_value = check_flag_value(flag="<flag-name>", user=user, default_value=default_value)
```
You can get `flag-name` from launchdarkly.
## Tracing Code
You can use tracing as follows:
```python
from src.external_interfaces.metrics.tracer import get_tracer
with get_tracer(module_name=__name__, span="<span-name>"):
# code to trace
```
Sign in with Wallet
Connect another wallet