Circular Dependency- a never ending Paradox in Software Development

sivakumarmanoharan — Sun, 23 Feb 2025 16:45:32 +0000

Have you ever thought of the paradoxes in life such as which came first, the egg or the chicken? In software also, there are such paradoxical problems that occur during development. One such paradox is the circular dependency due to imports. What is exactly a circular dependency and how we can avoid that dilemma to a machine so that it can execute it sequentially? Let us deep dive in those things with this post.

What is a circular dependency?

A circular dependency is similar to a race condition in the operating system thread execution. This is caused due to the imports happened in a program that are modularly separated. Let us see with an example.

In the above figure, the program is executed in a sequential way where the code pass the parameters to a function in B and then from B it is passing to another function in C. The return order will also be in the reverse direction.

But in circular dependency, if the function written in B requires C and if the function in C requires a function written in B, the machine gets confused on which to be executed first. As a result, it throws an error stating that circular imports could not be resolved.

# user.py
from fastapi import FastAPI
from pydantic import BaseModel
from post import Post

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    posts: list[Post]

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Fetch user and return
    pass

# post.py
from fastapi import FastAPI
from pydantic import BaseModel
from user import User

app = FastAPI()

class Post(BaseModel):
    id: int
    title: str
    author: User

@app.get("/posts/{post_id}")
async def get_post(post_id: int):
    # Fetch post and return
    pass

How to avoid circular dependencies?

Using ForwardRef in FastAPI applications:

# user.py
from fastapi import FastAPI
from pydantic import BaseModel, ForwardRef

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    posts: list[ForwardRef('Post')] = []

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Fetch user and return
    pass

# post.py
from fastapi import FastAPI
from pydantic import BaseModel, ForwardRef
from user import User

app = FastAPI()

class Post(BaseModel):
    id: int
    title: str
    author: ForwardRef('User')

Post.model_rebuild()
User.model_rebuild()

@app.get("/posts/{post_id}")
async def get_post(post_id: int):
    # Fetch post and return
    pass

Here, we are using Pydantic's ForwardRef class to define the models and utilizing model_rebuild() function to clear the circular dependencies happening by redefining the schemas during runtime execution.

Organization of modules and functions:

An evergreen proverb states that prevention is better than cure. Before circular dependency hits, it is better to organize the functions and define where it should be called. Proper calling of functions and proper organization of functions totally avoid such qualms to a machine and the execution happens smoothly.

Moving of shared code into a separate module:

If a function is required by two or other modules from different sources, it is better to move the common function into a separate module. This avoids race conditions within the program and allows better readability.

Conclusion

Thus, these are some of the methods through which we can solve and avoid such issues happening during development. What are the other methods we can use to solve this? Please ping in the comments section so that I can also learn and improve myself.

Thank you to all viewers for reading this article so patiently. I hope you would have derived something out of it!!

Caching in FastAPI: Unlocking High-Performance Development:

sivakumarmanoharan — Fri, 18 Oct 2024 01:14:18 +0000

In today’s digital world, every action—whether it’s swiping on a dating app or completing a purchase—relies on APIs working efficiently behind the scenes. As back-end developers, we know that every millisecond counts. But how can we make APIs respond faster? The answer lies in caching.

Caching is a technique that stores frequently accessed data in memory, allowing APIs to respond instantly instead of querying a slower database every time. Think of it like keeping key ingredients (salt, pepper, oil) on your kitchen countertop instead of fetching them from the pantry each time you cook—this saves time and makes the process more efficient. Similarly, caching reduces API response times by storing commonly requested data in a fast, accessible spot, like Redis.

Required Libraries to be installed

To connect with Redis Cache with FastAPI, we require the following libraries to be pre-installed.

pip install fastapi uvicorn aiocache pydantic

Pydantic is for creating database tables and structures. aiocache will perform asynchronous operations on Cache. uvicorn is responsible for the server running.

Redis Setup and Verification:

Setting up Redis directly in a Windows system is not possible at this juncture. Therefore, it has to be setup and run in Windows Subsystem for Linux. Instructions for installing WSL is given below

Install WSL | Microsoft Learn

Install Windows Subsystem for Linux with the command, wsl --install. Use a Bash terminal on your Windows machine run by your preferred Linux distribution - Ubuntu, Debian, SUSE, Kali, Fedora, Pengwin, Alpine, and more are available.

learn.microsoft.com

Post installing WSL, the following commands are required to install Redis

sudo apt update
sudo apt install redis-server
sudo systemctl start redis

To test Redis server connectivity, the following command is used

redis-cli

After this command, it will enter into a virtual terminal of port 6379. In that terminal, the redis commands can be typed and tested.

Setting Up the FastAPI Application

Let’s create a simple FastAPI app that retrieves user information and caches it for future requests. We will use Redis for storing cached responses.

Step 1: Define the Pydantic Model for User Data

We’ll use Pydantic to define our User model, which represents the structure of the API response.

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    age: int

Step 2: Create a Caching Decorator

To avoid repeating the caching logic for each endpoint, we’ll create a reusable caching decorator using the aiocache library. This decorator will attempt to retrieve the response from Redis before calling the actual function.

import json
from functools import wraps
from aiocache import Cache
from fastapi import HTTPException

def cache_response(ttl: int = 60, namespace: str = "main"):
    """
    Caching decorator for FastAPI endpoints.

    ttl: Time to live for the cache in seconds.
    namespace: Namespace for cache keys in Redis.
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            user_id = kwargs.get('user_id') or args[0]  # Assuming the user ID is the first argument
            cache_key = f"{namespace}:user:{user_id}"

            cache = Cache.REDIS(endpoint="localhost", port=6379, namespace=namespace)

            # Try to retrieve data from cache
            cached_value = await cache.get(cache_key)
            if cached_value:
                return json.loads(cached_value)  # Return cached data

            # Call the actual function if cache is not hit
            response = await func(*args, **kwargs)

            try:
                # Store the response in Redis with a TTL
                await cache.set(cache_key, json.dumps(response), ttl=ttl)
            except Exception as e:
                raise HTTPException(status_code=500, detail=f"Error caching data: {e}")

            return response
        return wrapper
    return decorator

Step 3: Implement a FastAPI Route for User Details

We’ll now implement a FastAPI route that retrieves user information based on a user ID. The response will be cached using Redis for faster access in subsequent requests.

from fastapi import FastAPI

app = FastAPI()

# Sample data representing users in a database
users_db = {
    1: {"id": 1, "name": "Alice", "email": "alice@example.com", "age": 25},
    2: {"id": 2, "name": "Bob", "email": "bob@example.com", "age": 30},
    3: {"id": 3, "name": "Charlie", "email": "charlie@example.com", "age": 22},
}

@app.get("/users/{user_id}")
@cache_response(ttl=120, namespace="users")
async def get_user_details(user_id: int):
    # Simulate a database call by retrieving data from users_db
    user = users_db.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

Step 4: Run the Application

Start your FastAPI application by running:

uvicorn main:app --reload

Now, you can test the API by fetching user details via:

http://127.0.0.1:8000/users/1

The first request will fetch the data from the users_db, but subsequent requests will retrieve the data from Redis.

Testing the Cache

You can verify the cache by inspecting the keys stored in Redis. Open the Redis CLI:

redis-cli
KEYS *

You will get all keys that have been stored in the Redis till TTL.

How Caching Works in This Example

First Request

: When the user data is requested for the first time, the API fetches it from the database (users_db) and stores the result in Redis with a time-to-live (TTL) of 120 seconds.

Subsequent Requests:

Any subsequent requests for the same user within the TTL period are served directly from Redis, making the response faster and reducing the load on the database.

TTL (Time to Live):

After 120 seconds, the cache entry expires, and the data is fetched from the database again on the next request, refreshing the cache.

Conclusion

In this tutorial, we’ve demonstrated how to implement Redis caching in a FastAPI application using a simple user details example. By caching API responses, you can significantly improve the performance of your application, particularly for data that doesn't change frequently.

Please do upvote and share if you find this article useful.

DEV Community: sivakumarmanoharan

Circular Dependency- a never ending Paradox in Software Development

What is a circular dependency?

How to avoid circular dependencies?

Conclusion

Caching in FastAPI: Unlocking High-Performance Development:

Required Libraries to be installed

Redis Setup and Verification:

Install WSL | Microsoft Learn

Setting Up the FastAPI Application

Step 1: Define the Pydantic Model for User Data

Step 2: Create a Caching Decorator

Step 3: Implement a FastAPI Route for User Details

Step 4: Run the Application

Testing the Cache

How Caching Works in This Example

First Request

Subsequent Requests:

TTL (Time to Live):

Conclusion