Skip to content

Commit 0414bea

Browse files
ochuiochui
authored
Merge pull request #8 from flowdacity/feat/migrate-from-config-file
Refactors config to accept mapping, removes INI support
2 parents b37ac71 + 2968d71 commit 0414bea

File tree

9 files changed

+287
-263
lines changed

9 files changed

+287
-263
lines changed

‎README.md‎

Lines changed: 40 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -33,22 +33,25 @@ pip install -e .
3333

3434
## Configuration
3535

36-
FQ reads a simple INI config file. Intervals are in milliseconds.
37-
```
38-
[fq]
39-
job_expire_interval : 5000
40-
job_requeue_interval : 5000
41-
default_job_requeue_limit : -1 ; -1 retries forever, 0 means no retries
42-
43-
[redis]
44-
db : 0
45-
key_prefix : queue_server
46-
conn_type : tcp_sock ; or unix_sock
47-
host : 127.0.0.1
48-
port : 6379
49-
password :
50-
clustered : false
51-
unix_socket_path : /tmp/redis.sock
36+
FQ accepts a simple config mapping. Intervals are in milliseconds.
37+
```python
38+
config = {
39+
"fq": {
40+
"job_expire_interval": 5000,
41+
"job_requeue_interval": 5000,
42+
"default_job_requeue_limit": -1, # -1 retries forever, 0 means no retries
43+
},
44+
"redis": {
45+
"db": 0,
46+
"key_prefix": "queue_server",
47+
"conn_type": "tcp_sock", # or "unix_sock"
48+
"host": "127.0.0.1",
49+
"port": 6379,
50+
"password": "",
51+
"clustered": False,
52+
"unix_socket_path": "/tmp/redis.sock",
53+
},
54+
}
5255
```
5356

5457
> If you connect via Unix sockets, uncomment the `unixsocket` lines in your `redis.conf`:
@@ -66,8 +69,26 @@ from fq import FQ
6669
6770
6871
async def main():
69-
fq = FQ("config.conf")
70-
await fq.initialize() # load config, connect to Redis, register Lua scripts
72+
config = {
73+
"fq": {
74+
"job_expire_interval": 5000,
75+
"job_requeue_interval": 5000,
76+
"default_job_requeue_limit": -1,
77+
},
78+
"redis": {
79+
"db": 0,
80+
"key_prefix": "queue_server",
81+
"conn_type": "tcp_sock",
82+
"host": "127.0.0.1",
83+
"port": 6379,
84+
"password": "",
85+
"clustered": False,
86+
"unix_socket_path": "/tmp/redis.sock",
87+
},
88+
}
89+
90+
fq = FQ(config)
91+
await fq.initialize() # connect to Redis and register Lua scripts
7192
7293
job_id = str(uuid.uuid4())
7394
await fq.enqueue(
@@ -102,7 +123,7 @@ Common operations:
102123

103124
## Development
104125

105-
- Start Redis for local development: `make redis-up` (binds to `localhost:6379`; matches `tests/test.conf`).
126+
- Start Redis for local development: `make redis-up` (binds to `localhost:6379`).
106127
- Run the suite: `make test` (automatically starts and tears down Redis).
107128
- Build a wheel: `make build`
108129
- Install/uninstall from the build: `make install` / `make uninstall`

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ dynamic = ["version"]
44
description = "A simple Redis-based job queue system."
55
readme = "README.md"
66
license = {text = "MIT"}
7-
authors = [{name = "Ochui Princewill", email = "ochui@flowdacity.com"}]
7+
authors = [{name = "Flowdacity Development Team", email = "admin@flowdacity.com"}]
88
requires-python = ">=3.12"
99
dependencies = [
1010
"msgpack>=1.1.2",

‎src/fq/default.conf‎

Lines changed: 0 additions & 14 deletions
This file was deleted.

‎src/fq/queue.py‎

Lines changed: 119 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
import asyncio
66
import os
7-
import configparser
7+
from collections.abc import Mapping
88

99
from redis.asyncio import Redis
1010
from redis.asyncio.cluster import RedisCluster
@@ -25,51 +25,144 @@ class FQ(object):
2525
"""The FQ object is the core of this queue.
2626
FQ does the following.
2727
28-
1. Accepts a configuration file.
28+
1. Accepts structured configuration.
2929
2. Initializes the queue.
3030
3. Exposes functions to interact with the queue.
3131
"""
3232

33-
def __init__(self, config_path):
33+
def __init__(self, config):
3434
"""Construct a FQ object by doing the following.
35-
1. Read the configuration path.
36-
2. Load the config.
35+
1. Store the queue configuration.
36+
2. Validate the config shape.
3737
"""
38-
self.config_path = config_path
39-
self._load_config()
4038
self._r = None # redis client placeholder
39+
if not isinstance(config, Mapping):
40+
raise FQException("Config must be a mapping with redis and fq sections")
41+
42+
normalized = {}
43+
for section_name, section_values in config.items():
44+
if not isinstance(section_values, Mapping):
45+
raise FQException(
46+
"Config section '%s' must be a mapping" % section_name
47+
)
48+
49+
normalized[str(section_name)] = {
50+
str(option): value for option, value in section_values.items()
51+
}
52+
53+
if "redis" not in normalized or "fq" not in normalized:
54+
raise FQException("Config missing required sections: redis, fq")
55+
56+
redis_config = normalized["redis"]
57+
fq_config = normalized["fq"]
58+
59+
if "key_prefix" not in redis_config:
60+
raise FQException("Missing config: redis.key_prefix")
61+
if not isinstance(redis_config["key_prefix"], str) or not redis_config[
62+
"key_prefix"
63+
]:
64+
raise FQException(
65+
"Invalid config: redis.key_prefix must be a non-empty string"
66+
)
67+
68+
if "conn_type" not in redis_config:
69+
raise FQException("Missing config: redis.conn_type")
70+
if redis_config["conn_type"] not in {"tcp_sock", "unix_sock"}:
71+
raise FQException(
72+
"Invalid config: redis.conn_type must be 'tcp_sock' or 'unix_sock'"
73+
)
74+
75+
if "db" not in redis_config:
76+
raise FQException("Missing config: redis.db")
77+
if isinstance(redis_config["db"], bool) or not isinstance(
78+
redis_config["db"], int
79+
):
80+
raise FQException("Invalid config: redis.db must be an integer")
81+
82+
if "job_expire_interval" not in fq_config:
83+
raise FQException("Missing config: fq.job_expire_interval")
84+
if not is_valid_interval(fq_config["job_expire_interval"]):
85+
raise FQException(
86+
"Invalid config: fq.job_expire_interval must be a positive integer"
87+
)
88+
89+
if "job_requeue_interval" not in fq_config:
90+
raise FQException("Missing config: fq.job_requeue_interval")
91+
if not is_valid_interval(fq_config["job_requeue_interval"]):
92+
raise FQException(
93+
"Invalid config: fq.job_requeue_interval must be a positive integer"
94+
)
95+
96+
if "default_job_requeue_limit" not in fq_config:
97+
raise FQException("Missing config: fq.default_job_requeue_limit")
98+
if not is_valid_requeue_limit(fq_config["default_job_requeue_limit"]):
99+
raise FQException(
100+
"Invalid config: fq.default_job_requeue_limit must be an integer >= -1"
101+
)
102+
103+
if redis_config["conn_type"] == "unix_sock":
104+
if "unix_socket_path" not in redis_config:
105+
raise FQException("Missing config: redis.unix_socket_path")
106+
if not isinstance(redis_config["unix_socket_path"], str) or not redis_config[
107+
"unix_socket_path"
108+
]:
109+
raise FQException(
110+
"Invalid config: redis.unix_socket_path must be a non-empty string"
111+
)
112+
113+
if redis_config["conn_type"] == "tcp_sock":
114+
if "host" not in redis_config:
115+
raise FQException("Missing config: redis.host")
116+
if not isinstance(redis_config["host"], str) or not redis_config["host"]:
117+
raise FQException(
118+
"Invalid config: redis.host must be a non-empty string"
119+
)
120+
121+
if "port" not in redis_config:
122+
raise FQException("Missing config: redis.port")
123+
if isinstance(redis_config["port"], bool) or not isinstance(
124+
redis_config["port"], int
125+
):
126+
raise FQException("Invalid config: redis.port must be an integer")
127+
128+
if "clustered" in redis_config and not isinstance(
129+
redis_config["clustered"], bool
130+
):
131+
raise FQException("Invalid config: redis.clustered must be a boolean")
132+
133+
if "password" in redis_config and redis_config["password"] is not None:
134+
if not isinstance(redis_config["password"], str):
135+
raise FQException("Invalid config: redis.password must be a string")
136+
137+
self.config = normalized
41138

42139
async def initialize(self):
43140
"""Async initializer to set up redis and lua scripts."""
44-
await self._initialize()
45-
46-
async def _initialize(self):
47-
"""Read the FQ configuration and set up redis + Lua scripts."""
141+
fq_config = self.config["fq"]
142+
redis_config = self.config["redis"]
48143

49-
self._key_prefix = self._config.get("redis", "key_prefix")
50-
self._job_expire_interval = int(self._config.get("fq", "job_expire_interval"))
51-
self._default_job_requeue_limit = int(
52-
self._config.get("fq", "default_job_requeue_limit")
53-
)
144+
self._key_prefix = redis_config["key_prefix"]
145+
self._job_expire_interval = int(fq_config["job_expire_interval"])
146+
self._default_job_requeue_limit = int(fq_config["default_job_requeue_limit"])
54147

55-
redis_connection_type = self._config.get("redis", "conn_type")
56-
db = self._config.get("redis", "db")
148+
redis_connection_type = redis_config["conn_type"]
149+
db = redis_config["db"]
57150

58151
if redis_connection_type == "unix_sock":
59152
self._r = Redis(
60153
db=db,
61-
unix_socket_path=self._config.get("redis", "unix_socket_path"),
154+
unix_socket_path=redis_config["unix_socket_path"],
62155
)
63156
elif redis_connection_type == "tcp_sock":
64157
isclustered = False
65-
if self._config.has_option("redis", "clustered"):
66-
isclustered = self._config.getboolean("redis", "clustered")
158+
if "clustered" in redis_config:
159+
isclustered = redis_config["clustered"]
67160

68161
if isclustered:
69162
startup_nodes = [
70163
{
71-
"host": self._config.get("redis", "host"),
72-
"port": int(self._config.get("redis", "port")),
164+
"host": redis_config["host"],
165+
"port": int(redis_config["port"]),
73166
}
74167
]
75168
self._r = RedisCluster(
@@ -80,9 +173,9 @@ async def _initialize(self):
80173
else:
81174
self._r = Redis(
82175
db=db,
83-
host=self._config.get("redis", "host"),
84-
port=int(self._config.get("redis", "port")),
85-
password=self._config.get("redis", "password"),
176+
host=redis_config["host"],
177+
port=int(redis_config["port"]),
178+
password=redis_config.get("password"),
86179
)
87180
else:
88181
raise FQException("Unknown redis conn_type: %s" % redis_connection_type)
@@ -107,36 +200,9 @@ async def _validate_redis_connection(self):
107200
if result is False:
108201
raise FQException("Failed to connect to Redis: ping returned False")
109202

110-
def _load_config(self):
111-
"""Read the configuration file and load it into memory."""
112-
if not os.path.isfile(self.config_path):
113-
raise FQException("Config file not found: %s" % self.config_path)
114-
115-
self._config = configparser.ConfigParser()
116-
read_files = self._config.read(self.config_path)
117-
118-
if not read_files:
119-
raise FQException("Unable to read config file: %s" % self.config_path)
120-
121-
if not self._config.has_section("redis") or not self._config.has_section(
122-
"fq"
123-
):
124-
raise FQException(
125-
"Config file missing required sections: redis, fq (path: %s)"
126-
% self.config_path
127-
)
128-
129203
def redis_client(self):
130204
return self._r
131205

132-
def reload_config(self, config_path=None):
133-
"""Reload the configuration from the new config file if provided
134-
else reload the current config file.
135-
"""
136-
if config_path:
137-
self.config_path = config_path
138-
self._load_config()
139-
140206
def _load_lua_scripts(self):
141207
"""Loads all lua scripts required by FQ."""
142208
# load lua scripts

‎tests/config.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
# -*- coding: utf-8 -*-
2+
3+
from copy import deepcopy
4+
5+
6+
TEST_CONFIG = {
7+
"fq": {
8+
"job_expire_interval": 5000,
9+
"job_requeue_interval": 5000,
10+
"default_job_requeue_limit": -1,
11+
},
12+
"redis": {
13+
"db": 0,
14+
"key_prefix": "test_fq",
15+
"conn_type": "tcp_sock",
16+
"unix_socket_path": "/tmp/redis.sock",
17+
"port": 6379,
18+
"host": "127.0.0.1",
19+
"clustered": False,
20+
"password": "",
21+
},
22+
}
23+
24+
25+
def build_test_config(**section_overrides):
26+
config = deepcopy(TEST_CONFIG)
27+
for section_name, overrides in section_overrides.items():
28+
config.setdefault(section_name, {})
29+
config[section_name].update(overrides)
30+
return config

‎tests/test.conf‎

Lines changed: 0 additions & 17 deletions
This file was deleted.

0 commit comments

Comments
 (0)