The Singleton pattern is one of the most well-known design patterns in software development. It ensures that a class has only one instance and provides a global point of access to it. In synchronous Python, this is fairly straightforward. But when you step into the world of asynchronous programming, things get a bit trickier.
In this blog post, we’ll start with the basics of the Singleton pattern, then move into the async/await world, and finally explore advanced usage patterns and real-life backend scenarios. We’ll also highlight common pitfalls and how to avoid them.
The Basics: What is a Singleton?
A Singleton is a class that can only have one instance throughout the lifetime of an application. In Python, this is usually implemented by overriding the __new__
method or using metaclasses.
Classic Singleton (Synchronous)
class Singleton:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
This pattern works fine when everything is synchronous. But what happens if you need to initialize the singleton asynchronously, like when making a database connection?
Enter Asynchronous Python
Asynchronous Python allows you to write non-blocking code using async
and await
. It’s perfect for I/O-bound tasks like API calls, database queries, or file operations. But here’s the catch: you can’t await
inside __new__
or __init__
, because they're not async
functions. So how do you create an async-aware Singleton?
Async Singleton Pattern
Let’s say we want to create a Singleton that manages a database connection. The connection setup is asynchronous. For example using an async PostgreSQL client.
Example 1: Async Database Connector Singleton
import asyncio
class AsyncDBConnector:
_instance = None
_lock = asyncio.Lock()
def __init__(self):
self.connection = None
@classmethod
async def get_instance(cls):
if cls._instance is None:
async with cls._lock:
if cls._instance is None:
instance = cls()
await instance._connect()
cls._instance = instance
return cls._instance
async def _connect(self):
# Simulate async DB connection
await asyncio.sleep(1)
self.connection = "Connected to DB"
Why it works:
- We use a class method
get_instance
to create the instance. - We wrap it with a lock to ensure thread safety in a multi-tasking environment.
- The
_connect
method is async and performs the setup.
Usage:
async def main():
db = await AsyncDBConnector.get_instance()
print(db.connection)
asyncio.run(main())
Example 2: Singleton Config Loader
Let’s build a singleton config loader that reads secrets from an external async API like AWS Secrets Manager.
class ConfigManager:
_instance = None
_lock = asyncio.Lock()
def __init__(self):
self.config = {}
@classmethod
async def get_instance(cls):
if cls._instance is None:
async with cls._lock:
if cls._instance is None:
instance = cls()
await instance._load_config()
cls._instance = instance
return cls._instance
async def _load_config(self):
await asyncio.sleep(0.5) # Simulate network request
self.config = {
"DATABASE_URL": "postgresql://user:pass@host/db",
"API_KEY": "abc123xyz"
}
# Usage
async def main():
config = await ConfigManager.get_instance()
print(config.config["API_KEY"])
asyncio.run(main())
Using Async Singleton in FastAPI
FastAPI is one of the most popular async web frameworks in Python. It thrives on non-blocking I/O and dependency injection. Let’s see how you can integrate an async Singleton pattern into a FastAPI app.
Use Case: Shared Redis Client or DB Connection
Imagine you have a Redis connection that should be initialized once and reused across all endpoints.
from fastapi import FastAPI, Depends
import asyncio
class RedisClient:
_instance = None
_lock = asyncio.Lock()
def __init__(self):
self.connection = None
@classmethod
async def get_instance(cls):
if cls._instance is None:
async with cls._lock:
if cls._instance is None:
instance = cls()
await instance._connect()
cls._instance = instance
return cls._instance
async def _connect(self):
await asyncio.sleep(1)
self.connection = "Simulated Redis Connection"
# Dependency wrapper
async def get_redis_client() -> RedisClient:
return await RedisClient.get_instance()
# FastAPI app
app = FastAPI()
@app.get("/status")
async def status(redis: RedisClient = Depends(get_redis_client)):
return {"status": "ok", "redis": redis.connection}
-
get_redis_client()
is declared as an async dependency. - The Singleton instance is resolved on demand and shared afterward.
- This is perfect for Redis, PostgreSQL, and custom HTTP clients like
aiohttp
orhttpx
.
Common Pitfalls
1. Trying to await
in __init__
This will fail:
class BadExample:
def __init__(self):
await self.setup() # SyntaxError
You can't await
inside __init__
. Use a separate async setup method.
2. Not Using a Lock
If multiple coroutines call get_instance()
at the same time, you might end up with multiple instances. Always protect instantiation with an asyncio.Lock()
to ensure thread safety.
A Few Tips
Make Singleton Test-Friendly
Sometimes it’s useful to reset or override the Singleton during testing.
@classmethod
def reset_instance(cls):
cls._instance = None
Avoid Global State
Even with Singletons, be cautious not to abuse them as global variables. It’s best to inject them where needed like as constructor args.
Lazy Initialization
Delay the heavy work until it's really needed. This keeps startup time low.
Wrapping Up
Singletons in async Python aren’t hard, but they do require you to rethink how and when your object is created. The key takeaway is: move all await
logic out of __init__
and into an async method, and make sure to guard instance creation with an asyncio.Lock()
.
Used right, an async Singleton can be a great tool for managing shared resources like DB connections, config loaders, and API clients in modern backend applications.
The original post is here.
Top comments (0)