Asyncpg_pool

asyncpg_pool

import asyncio
import asyncpg

import functools
import time
from typing import Callable, Any


def async_timed():
    def wrapper(func: Callable) -> Callable:
        @functools.wraps(func)
        async def wrapped(*args, **kwargs) -> Any:
            print(f'starting {func} with args {args} {kwargs}')
            start = time.time()
            try:
                return await func(*args, **kwargs)
            finally:
                end = time.time()
                total = end - start
                print(f'finished {func} in {total:.4f} second(s)')

        return wrapped

    return wrapper


product_query = \
    """
SELECT
p.product_id,
p.product_name,
p.brand_id,
s.sku_id,
pc.product_color_name,
ps.product_size_name
FROM product as p
JOIN sku as s on s.product_id = p.product_id
JOIN product_color as pc on pc.product_color_id = s.product_color_id
JOIN product_size as ps on ps.product_size_id = s.product_size_id
WHERE p.product_id = 100"""


async def query_product(pool):
    async with pool.acquire() as connection:
        return await connection.fetchrow(product_query)


@async_timed()
async def query_products_synchronously(pool, queries):
    return [await query_product(pool) for _ in range(queries)]


@async_timed()
async def query_products_concurrently(pool, queries):
    queries = [query_product(pool) for _ in range(queries)]
    return await asyncio.gather(*queries)

class Database:
    def __init__(self):
        # self.pool = None
        pass

    async def get_pool(self):
        return await asyncpg.create_pool(host='127.0.0.1',
                                         port=5432,
                                         user='postgres',
                                         password='shopback',
                                         database='simulation',
                                         min_size=10,
                                         max_size=30)


async def main():
    db = Database()
    pool = await db.get_pool()
    await query_products_synchronously(pool, 10000)
    await query_products_concurrently(pool, 10000)


asyncio.run(main())