Skip to content

Commit 01e32d8

Browse files
fernaperfernaper
committed
🎉 v0.3.0
✨ Added support for environment variables
1 parent dade549 commit 01e32d8

6 files changed

Lines changed: 209 additions & 15 deletions

File tree

‎README.md‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -217,9 +217,27 @@ if __name__ == '__main__':
217217
main()
218218
```
219219
220+
## Configuration
221+
222+
You can configure FlowGrid by setting environment variables. Here are all keys and their default values:
223+
224+
- FLOWGRID_CELERY_BROKER_URL=amqp://guest:guest@localhost:5672//
225+
- FLOWGRID_CELERY_RESULT_BACKEND=redis://localhost:6380/0
226+
- FLOWGRID_SERIALIZER=json
227+
- FLOWGRID_TASK_SERIALIZER=json
228+
- FLOWGRID_RESULT_SERIALIZER=json
229+
- FLOWGRID_ACCEPT_CONTENT=json
230+
- FLOWGRID_TIMEZONE=UTC
231+
- FLOWGRID_ENABLE_UTC=True
232+
233+
234+
Serializer, task serializer, result serializer, and accept content can be set to `json`, `pickle` or `msgpac`.
235+
Serializer is a global setting that will be used for all serializers if not set.
236+
237+
The timezone can be set to any valid timezone. The `FLOWGRID_ENABLE_UTC` variable can be set to `True` or `False` (case-insensitive).
238+
220239
## Future Enhancements
221240
222-
- **PyPI Release**: Soon, FlowGrid will be available for installation via pip.
223241
- **Extended Documentation**: More detailed documentation and examples will be added as the project evolves.
224242
- **Support chaining tasks without waiting**: Currently, you have to wait for a task to finish before chaining another task. You can check the example `examples/06-chaining.py` to see how to chain tasks. With the next solution it will look in the exact same way but if you use `fg.launch` the response will be instant and you can return the task to the user from the beginning.
225243
- **Support task sequence without relationship**: Right now you can only chain tasks just by including them as paramteres, but what if the response of a task is not needed for the next one. We will create an interface called: `fg.sequence` that will allow you to define a sequence of tasks that will be executed in order.

‎examples/08-pickle-serializer.py‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
import os
2+
import pathlib
3+
4+
# os.environ['FLOWGRID_TASK_SERIALIZER'] = 'pickle'
5+
# os.environ['FLOWGRID_RESULT_SERIALIZER'] = 'pickle'
6+
# os.environ['FLOWGRID_ACCEPT_CONTENT'] = 'pickle'
7+
# Equivalent to:
8+
os.environ['FLOWGRID_SERIALIZER'] = 'pickle'
9+
10+
from flowgrid import FlowGrid # noqa E402
11+
12+
fg = FlowGrid()
13+
14+
15+
@fg.task
16+
def dummy_example(path: pathlib.Path) -> pathlib.Path:
17+
'''
18+
A dummy example of a non-JSON serializable task, since
19+
pathlib.Path is not serializable by JSON.
20+
'''
21+
return path / 'new_file.txt'
22+
23+
24+
def main():
25+
result = fg.wait(
26+
dummy_example(pathlib.Path.cwd())
27+
)
28+
print(result)
29+
# Expected: /path/to/current/directory/new_file.txt
30+
31+
32+
if __name__ == '__main__':
33+
main()

‎flowgrid/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22
from .celery_app import make_celery
33

44
__all__ = ['FlowGrid', 'Task', 'make_celery', '__version__']
5-
__version__ = '0.2.1'
5+
__version__ = '0.3.0'

‎flowgrid/celery_app.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,13 @@
44

55

66
def make_celery(name: str = 'FlowGrid') -> Celery:
7+
config = Config.from_env()
78
celery = Celery(
89
name,
9-
broker=Config.CELERY_BROKER_URL,
10-
backend=Config.CELERY_RESULT_BACKEND,
10+
broker=config.celery_broker_url,
11+
backend=config.celery_result_backend,
1112
)
12-
celery.conf.update(Config.CELERY_CONFIG)
13+
celery.conf.update(config.celery_config)
1314
return celery
1415

1516

‎flowgrid/config.py‎

Lines changed: 151 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,152 @@
1+
import os
2+
from typing import List, Literal, TypedDict, Optional
3+
4+
5+
class CeleryConfig(TypedDict):
6+
task_serializer: Literal['json', 'pickle', 'msgpack']
7+
result_serializer: Literal['json', 'pickle', 'msgpack']
8+
accept_content: List[Literal['json', 'pickle', 'msgpack']]
9+
timezone: str
10+
enable_utc: bool
11+
12+
13+
VALID_SERIALIZERS: List[
14+
Literal['json', 'pickle', 'msgpack']
15+
] = ['json', 'pickle', 'msgpack']
16+
17+
118
class Config:
2-
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
3-
CELERY_RESULT_BACKEND = 'redis://localhost:6380/0'
4-
CELERY_CONFIG = {
5-
'task_serializer': 'json',
6-
'result_serializer': 'json',
7-
'accept_content': ['json'],
8-
'timezone': 'UTC',
9-
'enable_utc': True,
10-
}
19+
def __init__(
20+
self,
21+
celery_broker_url: Optional[str] = None,
22+
celery_result_backend: Optional[str] = None,
23+
celery_config: Optional[CeleryConfig] = None,
24+
):
25+
"""
26+
Initialize configuration with optional overrides.
27+
28+
Args:
29+
celery_broker_url: Celery broker URL. Defaults to environment
30+
variable or 'amqp://guest:guest@localhost:5672//'.
31+
celery_result_backend: Celery result backend. Defaults to
32+
environment variable or 'redis://localhost:6380/0'.
33+
celery_config: Celery configuration dictionary. Defaults to
34+
environment variable or a dictionary with the following keys
35+
and values:
36+
{
37+
'task_serializer': 'json',
38+
'result_serializer': 'json',
39+
'accept_content': ['json'],
40+
'timezone': 'UTC',
41+
'enable_utc': True
42+
}
43+
"""
44+
# Global serializer from environment variable
45+
global_serializer = next(
46+
(
47+
serializer for serializer in VALID_SERIALIZERS
48+
if serializer == os.getenv('FLOWGRID_SERIALIZER', '')
49+
),
50+
''
51+
)
52+
53+
# Celery Broker URL
54+
self.celery_broker_url = celery_broker_url or os.getenv(
55+
'FLOWGRID_CELERY_BROKER_URL',
56+
'amqp://guest:guest@localhost:5672//',
57+
)
58+
59+
# Celery Result Backend
60+
self.celery_result_backend = celery_result_backend or os.getenv(
61+
'FLOWGRID_CELERY_RESULT_BACKEND',
62+
'redis://localhost:6380/0',
63+
)
64+
65+
# Celery Configuration
66+
if celery_config is None:
67+
self.celery_config: CeleryConfig = {
68+
'task_serializer': next(
69+
(
70+
serializer for serializer in VALID_SERIALIZERS
71+
if serializer == os.getenv('FLOWGRID_TASK_SERIALIZER', '') # noqa E501
72+
),
73+
global_serializer or 'json'
74+
),
75+
'result_serializer': next(
76+
(
77+
serializer for serializer in VALID_SERIALIZERS
78+
if serializer == os.getenv('FLOWGRID_RESULT_SERIALIZER', '') # noqa E501
79+
),
80+
global_serializer or 'json'
81+
),
82+
'accept_content': [
83+
next(
84+
(
85+
serializer for serializer in VALID_SERIALIZERS
86+
if serializer == os.getenv('FLOWGRID_ACCEPT_CONTENT', '') # noqa E501
87+
),
88+
global_serializer or 'json'
89+
)
90+
],
91+
'timezone': os.getenv('FLOWGRID_TIMEZONE', 'UTC'),
92+
'enable_utc': os.getenv(
93+
'FLOWGRID_ENABLE_UTC',
94+
'True',
95+
).lower() == 'true'
96+
}
97+
else:
98+
self.celery_config = celery_config
99+
100+
self.validate()
101+
102+
def validate(self) -> None:
103+
"""
104+
Validate the configuration values.
105+
106+
Raises:
107+
ValueError: If any configuration value is invalid.
108+
"""
109+
# Validate Celery broker URL
110+
if not self.celery_broker_url.startswith(('amqp://', 'amqps://')):
111+
raise ValueError(
112+
f'Invalid Celery broker URL: {self.celery_broker_url}'
113+
)
114+
115+
# Validate Celery result backend
116+
if not self.celery_result_backend.startswith(
117+
('redis://', 'rediss://')
118+
):
119+
raise ValueError(
120+
f'Invalid Celery result backend: {self.celery_result_backend}'
121+
)
122+
123+
# Validate Celery config
124+
if self.celery_config['task_serializer'] not in VALID_SERIALIZERS:
125+
raise ValueError(
126+
f'Invalid task serializer: {self.celery_config['task_serializer']}' # noqa E501
127+
)
128+
129+
if self.celery_config['result_serializer'] not in VALID_SERIALIZERS:
130+
raise ValueError(
131+
f'Invalid result serializer: {self.celery_config['result_serializer']}' # noqa E501
132+
)
133+
134+
# Validate accept_content
135+
if not all(
136+
content in VALID_SERIALIZERS
137+
for content in self.celery_config['accept_content']
138+
):
139+
raise ValueError(
140+
f'Invalid accept_content: {self.celery_config['accept_content']}' # noqa E501
141+
)
142+
143+
@classmethod
144+
def from_env(cls) -> 'Config':
145+
"""
146+
Create a configuration instance from environment variables.
147+
148+
Returns:
149+
Config: A new configuration instance loaded from environment
150+
variables.
151+
"""
152+
return cls()

‎setup.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
try:
33
from flowgrid import __version__
44
except ImportError:
5-
__version__ = '0.2.1'
5+
__version__ = '0.3.0'
66

77
if __name__ == '__main__':
88
with open('README.md', 'r', encoding='utf8') as f:

0 commit comments

Comments
 (0)