Create an arq RedisSettings instance from a redis: URI (for example, as used in dokku or similar scenarios)

README.md

Populate an arq.connections.RedisSettings instance from a redis: connection URL string.

Useful for populating arq WorkerSettings etc.

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

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
    

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

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