Skip to content

Instantly share code, notes, and snippets.

@inactivist

inactivist/README.md

Last active March 21, 2024 21:58
Show Gist options
  • Save inactivist/d97aac848cc1bde847592905f62c3b6d to your computer and use it in GitHub Desktop.
Save inactivist/d97aac848cc1bde847592905f62c3b6d to your computer and use it in GitHub Desktop.
Download ZIP
Create an arq RedisSettings instance from a redis: URI (for example, as used in dokku or similar scenarios)
Raw
README.md
Raw
redis_settings_from_uri.py
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")
)
Raw
redis_settings_from_uri_example.py
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
Raw
test_redis_settings_from_uri.py
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"]
@python8787
Copy link

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment