Celery broker #12630
Celery broker #12630
Conversation
|
|
rasa/core/brokers/celery.py
Outdated
| task_instance: Celery = Celery("rasa_event_broker", broker=broker_url) | ||
| task_callable = partial(task_instance.send_task, task_name) | ||
| else: | ||
| task_instance = rasa.shared.utils.common.class_from_module_path(task_name) |
There was a problem hiding this comment.
This seems redundant as Celery requires URL to be present.
If broker_url is not present we need to issue RasaException to signal to the user that configuration they provided is not valid.
There was a problem hiding this comment.
@radovanZRasa, the idea here would be to allow users to have the choice to either use send_task or apply_async. If the user just uses apply_async, they would not need to pass it the broker_url. I did find that passing in the broker_url could be a bit of a hassle from the user's side as the user would need to pass it into the endpoint.yaml file via env variables. This is because the broker_url should contain sensitive auth info
If the user uses apply_async, they can use the broker_url that is already set functionally nearby their function. What do you think?
tests/core/test_broker.py
Outdated
|
|
||
|
|
||
| def test_celery_from_endpoint_config(): | ||
| cfg = read_endpoint_config( |
There was a problem hiding this comment.
Although we do have several YAML test configs from broker and other parts of Rasa we plan to move away from them as chasing them down is not as fast as having these as pytest fixtures.
Suggestion is to change this to:
@pytest.fixture
def celery_endpoint_config_string() -> str:
return """
event_broker:
type: celery
broker_url: rediss://:password@localhost:6379/0
task_name: tasks.event_logger
countdown: 2
priority: 1
"""
@pytest.fixture
def celery_endpoint_config(celery_endpoint_config_string: str) -> EndpointConfig:
yaml_content = read_yaml(celery_endpoint_config_string)
return EndpointConfig.from_dict(yaml_content)
def test_celery_from_endpoint_config(celery_endpoint_config: EndpointConfig):
actual = await EventBroker.create(celery_endpoint_config)
assert isinstance(actual, CeleryEventBroker)
rasa/core/brokers/celery.py
Outdated
| def __init__( | ||
| self, | ||
| broker_url: Optional[Text] = None, | ||
| task_name: Text = None, |
There was a problem hiding this comment.
Should be renamed to task_queue_name.
rasa/core/brokers/celery.py
Outdated
| self._task = self._get_task(broker_url, task_name) | ||
| self._task_kwargs: Dict[Text, Any] = kwargs | ||
|
|
||
| def _get_task( |
There was a problem hiding this comment.
There is no need to obfuscate send_task method from Celery.
You can just implement publish method as:
def self.celery_instance.send_task(self.task_queue_name, [event], **self._task_kwargs)This method can then be used to validate config and initialise client and thus renamed to match the purpose.
|
@radovanZRasa , please review |
Proposed changes:
Status (please check what you already did):
black(please check Readme for instructions)