sanic global singleton and unittest

sanic_singleton_unittest.md

global singleton

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

unittest

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")