Populate an arq.connections.RedisSettings
instance from a redis:
connection URL string.
Useful for populating arq WorkerSettings
etc.
Populate an arq.connections.RedisSettings
instance from a redis:
connection URL string.
Useful for populating arq WorkerSettings
etc.
from aioredis.util import parse_url | |
def redis_settings_from_uri(uri: str) -> RedisSettings: | |
address, options = parse_url(uri) | |
return RedisSettings( | |
host=address[0], port=address[1], password=options.get("password") | |
) |
class WorkerSettings: | |
# If using bound methods, they will have names prefixed with classname | |
# TwitterBotShell.process_mention, etc. | |
redis_settings = redis_settings_from_uri(uri=os.environ["REDIS_URL"]) | |
# etc | |
import pytest | |
from arq.connections import RedisSettings | |
from redis_settings_from_uri import redis_settings_from_uri | |
@pytest.mark.parametrize( | |
"uri,expected", | |
[ | |
("redis://localhost:6379", dict(host="localhost", port=6379, password=None)), | |
( | |
"redis://user:password@host:1234", | |
dict(host="host", port=1234, password="password"), | |
), | |
], | |
) | |
def test_redis_settings_from_uri(caplog, uri, expected): | |
rs: RedisSettings = redis_settings_from_uri(uri) | |
assert rs.host == expected["host"] | |
assert rs.port == expected["port"] | |
assert rs.password == expected["password"] | |
how to add arq worker timeout/expire. for example 'job_worker1' running on arq that should end up in certain time maybe after 10mints.
simply how to handle arq like rq... is their options to handle each worker-job