We are developing Web application based on sanic framework
Due to the complexity of business, we need to integrate http pool, redis/es pool, database connection pool, even if you're not going to use connection pool, you will need to singleton your connection to gain performance
Python
's async power is based on event driven loop, all the connections need to be integrate with the loop
Luckily, sanic
provides the following methods
sanic_app.register_listener(init_redis, 'before_server_start')
sanic_app.register_listener(close_redis, 'after_server_stop')
We can initialize our client, and attach it to the app
instance before the server start
async def init_redis(app: sanic.app, loop: asyncio.AbstractEventLoop) -> aioredis.pool.ConnectionsPool:
pool = await aioredis.create_pool(host, loop=loop)
app.redis_client = pool
But we found it not easy for us to write some module or sdk which will use the redis
client, we must pass the request.app.redis_client
client as parameter to initialize the instance
class ExportUtil(object):
def __init__(self, redis_cli: aioredis.pool.ConnectionsPool)
self.redis_cli = redis_cli
async def set_progress(key: str, data: str):
await self.redis_cli.execute("SETEX", key, 60, data)
async def get(self, request):
export_util = ExportUtil(request.app.redis_client)
await export_util.set_progress(key, data)
// ...
It implement our requirements, But it's not concise when we have many different places using redis
, We don't want to pass redis
and instantiate class ExportUtil
everywhere
We need a intuitive way
Let's try to singleton our redis
client globally
class RedisUtil(object):
client: aioredis.pool.ConnectionsPool = None
async def init_redis(app: sanic.app, loop: asyncio.AbstractEventLoop) -> aioredis.pool.ConnectionsPool:
RedisUtil.client = await aioredis.create_pool(host, loop=loop)
You can write your module in the following style now
from . import RedisUtil
async def set_progress(key: str, data: str):
await RedisUtil.client.execute("SETEX", key, 60, data)
async def get(self, request):
await set_progress(key, data)
Though redis
still need to be initialized when app
starts, app
and redis
has already been decoupled
Same function can be achieved by more concise code
You need pip install pytest-sanic
to make the following pytest
code works
But I found that pytest
will create new app
instance and new loop
among all test
cases, it means that the first test
case and second test
case can't shares same global redis instance
Beacause when executing the second test
case, loop is the second loop
, while the global redis client is attached to the first loop
We can do the following configuration
@pytest.fixture
def test_cli(loop, app, sanic_client):
return loop.run_until_complete(sanic_client(app))
async def test_1(self, test_cli):
resp = await test_cli.get("/api/record/")
data = await resp.json()
assert data[0]["id"] == 1
async def test_2(self, test_cli):
resp = await test_cli.put("/api/record/", ...)
data = await resp.json()
assert data[0]["id"] == 1
As long as the parameter includes test_cli
, when executing the test
function, new app
will be created and the specific before_server_start
will be executed, so redis
client will be initialized properly
But sometime the connection from my localhost to the remote redis
will take few seconds, I don't want the server starting process blocked by the redis connection, leave the handshake to the event loop
While in unittest
, I need the connection established before executing the test
, so I set a environment when starting pytest
The final version
class RedisUtil(object):
client: aioredis.pool.ConnectionsPool = None
async def init_connection(self) -> aioredis.RedisConnection:
res = await aioredis.create_pool(host)
setattr(RedisUtil, "client", res)
self.client = res
return res
async def init_redis(app: sanic.app, loop: asyncio.AbstractEventLoop):
app.redis_util = RedisUtil(loop)
if IS_UNITTEST:
await app.redis_util.init_connection()
else:
asyncio.ensure_future(app.redis_util.init_connection(), loop=loop)
What's more ? sanic
's Blueprint
is one time usage
cred_bp = sanic.Blueprint('item', url_prefix="/my_path")
def add_api_routes(app: sanic.Sanic):
app.blueprint(basic_bp)
api_bp = sanic.Blueprint.group(cred_bp, url_prefix="/api")
You will find your second test
case unable to find your path
You need to create new Blueprint
every time the app
created
def add_api_routes(app: sanic.Sanic):
cred_bp = sanic.Blueprint('item', url_prefix="/my_path")
app.blueprint(basic_bp)
api_bp = sanic.Blueprint.group(cred_bp, url_prefix="/api")