# Path: arq/connections.py
# async def create_pool(
#     settings_: RedisSettings = None,
#     *,
#     retry: int = 0,
#     job_serializer: Optional[Serializer] = None,
#     job_deserializer: Optional[Deserializer] = None,
#     default_queue_name: str = default_queue_name,
# ) -> ArqRedis:
#     """
#     Create a new redis pool, retrying up to ``conn_retries`` times if the connection fails.
# 
#     Returns a :class:`arq.connections.ArqRedis` instance, thus allowing job enqueuing.
#     """
#     settings: RedisSettings = RedisSettings() if settings_ is None else settings_
# 
#     assert not (
#         type(settings.host) is str and settings.sentinel
#     ), "str provided for 'host' but 'sentinel' is true; list of sentinels expected"
# 
#     if settings.sentinel:
# 
#         def pool_factory(*args: Any, **kwargs: Any) -> ArqRedis:
#             client = Sentinel(*args, sentinels=settings.host, ssl=settings.ssl, **kwargs)
#             return client.master_for(settings.sentinel_master, redis_class=ArqRedis)
# 
#     else:
#         pool_factory = functools.partial(
#             ArqRedis,
#             host=settings.host,
#             port=settings.port,
#             socket_connect_timeout=settings.conn_timeout,
#             ssl=settings.ssl,
#         )
# 
#     try:
#         pool = pool_factory(db=settings.database, password=settings.password, encoding='utf8')
#         pool.job_serializer = job_serializer
#         pool.job_deserializer = job_deserializer
#         pool.default_queue_name = default_queue_name
#         await pool.ping()
# 
#     except (ConnectionError, OSError, RedisError, asyncio.TimeoutError) as e:
#         if retry < settings.conn_retries:
#             logger.warning(
#                 'redis connection error %s:%s %s %s, %d retries remaining...',
#                 settings.host,
#                 settings.port,
#                 e.__class__.__name__,
#                 e,
#                 settings.conn_retries - retry,
#             )
#             await asyncio.sleep(settings.conn_retry_delay)
#         else:
#             raise
#     else:
#         if retry > 0:
#             logger.info('redis connection successful')
#         return pool
# 
#     # recursively attempt to create the pool outside the except block to avoid
#     # "During handling of the above exception..." madness
#     return await create_pool(
#         settings,
#         retry=retry + 1,
#         job_serializer=job_serializer,
#         job_deserializer=job_deserializer,
#         default_queue_name=default_queue_name,
#     )
# 
# Path: arq/connections.py
# class RedisSettings:
#     """
#     No-Op class used to hold redis connection redis_settings.
# 
#     Used by :func:`arq.connections.create_pool` and :class:`arq.worker.Worker`.
#     """
# 
#     host: Union[str, List[Tuple[str, int]]] = 'localhost'
#     port: int = 6379
#     database: int = 0
#     password: Optional[str] = None
#     ssl: Union[bool, None, SSLContext] = None
#     conn_timeout: int = 1
#     conn_retries: int = 5
#     conn_retry_delay: int = 1
# 
#     sentinel: bool = False
#     sentinel_master: str = 'mymaster'
# 
#     @classmethod
#     def from_dsn(cls, dsn: str) -> 'RedisSettings':
#         conf = urlparse(dsn)
#         assert conf.scheme in {'redis', 'rediss'}, 'invalid DSN scheme'
#         return RedisSettings(
#             host=conf.hostname or 'localhost',
#             port=conf.port or 6379,
#             ssl=conf.scheme == 'rediss',
#             password=conf.password,
#             database=int((conf.path or '0').strip('/')),
#         )
# 
#     def __repr__(self) -> str:
#         return 'RedisSettings({})'.format(', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()))


import asyncio
from aiohttp import ClientSession
from arq import create_pool
from arq.connections import RedisSettings

async def download_content(ctx, url):
    session: ClientSession = ctx['session']
    async with session.get(url) as response:
        content = await response.text()
        print(f'{url}: {content:.80}...')
    return len(content)

async def startup(ctx):
    ctx['session'] = ClientSession()


# Path: arq/connections.py
# async def create_pool(
#     settings_: RedisSettings = None,
#     *,
#     retry: int = 0,
#     job_serializer: Optional[Serializer] = None,
#     job_deserializer: Optional[Deserializer] = None,
#     default_queue_name: str = default_queue_name,
# ) -> ArqRedis:
#     """
#     Create a new redis pool, retrying up to ``conn_retries`` times if the connection fails.
# 
#     Returns a :class:`arq.connections.ArqRedis` instance, thus allowing job enqueuing.
#     """
#     settings: RedisSettings = RedisSettings() if settings_ is None else settings_
# 
#     assert not (
#         type(settings.host) is str and settings.sentinel
#     ), "str provided for 'host' but 'sentinel' is true; list of sentinels expected"
# 
#     if settings.sentinel:
# 
#         def pool_factory(*args: Any, **kwargs: Any) -> ArqRedis:
#             client = Sentinel(*args, sentinels=settings.host, ssl=settings.ssl, **kwargs)
#             return client.master_for(settings.sentinel_master, redis_class=ArqRedis)
# 
#     else:
#         pool_factory = functools.partial(
#             ArqRedis,
#             host=settings.host,
#             port=settings.port,
#             socket_connect_timeout=settings.conn_timeout,
#             ssl=settings.ssl,
#         )
# 
#     try:
#         pool = pool_factory(db=settings.database, password=settings.password, encoding='utf8')
#         pool.job_serializer = job_serializer
#         pool.job_deserializer = job_deserializer
#         pool.default_queue_name = default_queue_name
#         await pool.ping()
# 
#     except (ConnectionError, OSError, RedisError, asyncio.TimeoutError) as e:
#         if retry < settings.conn_retries:
#             logger.warning(
#                 'redis connection error %s:%s %s %s, %d retries remaining...',
#                 settings.host,
#                 settings.port,
#                 e.__class__.__name__,
#                 e,
#                 settings.conn_retries - retry,
#             )
#             await asyncio.sleep(settings.conn_retry_delay)
#         else:
#             raise
#     else:
#         if retry > 0:
#             logger.info('redis connection successful')
#         return pool
# 
#     # recursively attempt to create the pool outside the except block to avoid
#     # "During handling of the above exception..." madness
#     return await create_pool(
#         settings,
#         retry=retry + 1,
#         job_serializer=job_serializer,
#         job_deserializer=job_deserializer,
#         default_queue_name=default_queue_name,
#     )
# 
# Path: arq/connections.py
# class RedisSettings:
#     """
#     No-Op class used to hold redis connection redis_settings.
# 
#     Used by :func:`arq.connections.create_pool` and :class:`arq.worker.Worker`.
#     """
# 
#     host: Union[str, List[Tuple[str, int]]] = 'localhost'
#     port: int = 6379
#     database: int = 0
#     password: Optional[str] = None
#     ssl: Union[bool, None, SSLContext] = None
#     conn_timeout: int = 1
#     conn_retries: int = 5
#     conn_retry_delay: int = 1
# 
#     sentinel: bool = False
#     sentinel_master: str = 'mymaster'
# 
#     @classmethod
#     def from_dsn(cls, dsn: str) -> 'RedisSettings':
#         conf = urlparse(dsn)
#         assert conf.scheme in {'redis', 'rediss'}, 'invalid DSN scheme'
#         return RedisSettings(
#             host=conf.hostname or 'localhost',
#             port=conf.port or 6379,
#             ssl=conf.scheme == 'rediss',
#             password=conf.password,
#             database=int((conf.path or '0').strip('/')),
#         )
# 
#     def __repr__(self) -> str:
#         return 'RedisSettings({})'.format(', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()))


# Path: docs/examples/main_demo.py
import asyncio
from aiohttp import ClientSession
from arq import create_pool
from arq.connections import RedisSettings


async def download_content(ctx, url):
    session: ClientSession = ctx['session']
    async with session.get(url) as response:
        content = await response.text()
        print(f'{url}: {content:.80}...')
    return len(content)

async def startup(ctx):
    ctx['session'] = ClientSession()


async def shutdown(ctx):