Fastapi Essential Modules

Essential modules for developing applications with FastAPI. This project uses Python 3.10 as the environment and Poetry as the package manager.

Table of Contents

1. Quickstart
2. Migration
3. Logging
4. Caching
5. Job Scheduler
6. Rate-limiting

Quickstart

1. Open Terminal in this directory.

2. Initialize the virtual environment by command:

   python -m venv venv  # py -m venv venv

3. Use virtual environment by the activation command:

   venv\Scripts\activate # For Mac user: source bin/activate

4. Install Poetry and update required packages:

   pip install poetry
   
   poetry update

5. Run the application

   uvicorn app.main:app --reload

Migration

To add or initialize a module, you can use the following commands:

poetry add alembic  # pip install alembic

alembic init migrations

To create a new revision:

alembic revision --autogenerate -m "message"

To migrate:

alembic upgrade head

To undo the latest migration:

alembic downgrade -1

To roll back to a specific revision:

alembic downgrade <<revision_id>>

Logging

Everything is wrapped up in a function (app/core/logger.py) and it's just a matter of calling the function when initializing the application.

from app.core.logger import setup as setup_logging

@asynccontextmanager
async def lifespan(app: FastAPI):
    # start up
    setup_logging()
    yield
    # shut down
    pass

app = FastAPI(
    title=settings.PROJECT_NAME,
    lifespan=lifespan
)

See the example in app/api/utils.py.

Test with curl.

curl --request POST \
  --url 'http://localhost:8000/api/v1/utils/logs?text=This%20is%20log'

Caching

Install the Redis package.

poetry add redis   # pip install redis

Adapter for Redis: app\core\redis.py. To manage connections and disconnections to Redis

from app.core.redis import redis_client

@asynccontextmanager
async def lifespan(app: FastAPI):
    # start up
    await redis_client.connect(str(settings.REDIS_URL))
    yield
    # shut down
    await redis_client.disconnect()

app = FastAPI(
    title=settings.PROJECT_NAME,
    lifespan=lifespan
)

Caching module: app\core\cache.py

See the example in app\api\user.py

Test with curl.

# Insert 20000 users
curl --request POST \
  --url http://localhost:8000/api/v1/users/bulk/20000

# Get 20000 users
curl --request GET \
  --url 'http://localhost:8000/api/v1/users?limit=20000&skip=0'

Job Scheduler

Install the APScheduler package.

poetry add apscheduler   # pip install apscheduler

Scheduler module: app\core\scheduler.py

CRD APIs: app\api\jobs.py

Test with scripts in tests\job_scheduler folder.

python .\tests\job_scheduler\create_cron.py

python .\tests\job_scheduler\create_interval.py

python .\tests\job_scheduler\create_date_job.py

python .\tests\job_scheduler\get_list_jobs.py

python .\tests\job_scheduler\delete_job.py

Rate-limiting

Rate-limiting module: app/core/rate_limiter.py

Test APIs: app/api/utils.py

Test with scripts in tests\rate_limiting folder.

python .\tests\rate_limiting\test_rate_limit_1.py

python .\tests\rate_limiting\test_rate_limit_2.py