diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index f0796ff..4b183a0 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -81,10 +81,10 @@ class Gates(BaseActivity): self.logger.error(f"Filter {fil} not found") continue try: - if input_filter_functions[fil](data, config): + if input_filter_functions[fil](data, config['config']): self.logger.debug( f"Data not passed the input filter {fil}:{config}") - filter_output.append(config['POLICY']) + filter_output.append(config['policy']) except Exception as e: trace = traceback.format_exc() self.notification_handler.build_and_send_notification( @@ -139,7 +139,7 @@ class Gates(BaseActivity): continue try: if mlflow_response_filter_functions[fil](data, config): - filter_output.append(config['POLICY']) + filter_output.append(config['policy']) comments.append(data['content']['message']) self.notification_handler.build_and_send_notification( notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}", @@ -201,7 +201,7 @@ class Gates(BaseActivity): continue try: if mlflow_content_filter_functions[fil](data, config): - filter_output.append(config['POLICY']) + filter_output.append(config['policy']) self.notification_handler.build_and_send_notification( notification_id=f"{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}", message=f"Data not passed the content filter {fil}:{config}", diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index d98cebe..cdc6bf6 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -13,7 +13,7 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool bool: True if the specific columns have null values, False otherwise. """ return not data[ - data['variable'].isin(config['VARIABLES']) & data['value'].isna()].empty + data['variable'].isin(config['variables']) & data['value'].isna()].empty def filter_empty_data(data: DataFrame, _config: dict) -> bool: diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index f10716e..ae5e47c 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -32,10 +32,6 @@ async def main(): servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'), logger=logger, project_name=os.getenv('PROJECT_NAME', 'laborious'), - pipeline_name='-', - trigger_name='-', - model_name='-', - model='-' ) logger.info('Starting Activities...') @@ -60,7 +56,7 @@ async def main(): workers = [ Worker( temporal_client, - task_queue='predictions-queue', + task_queue='predictions_batch-queue', workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction], activities=[ diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index e4fc8cc..e3b38d8 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -205,7 +205,8 @@ class PredictionProcess(): { 'schema': schema, 'table_name': table_name, - 'model_id': model_id + 'model': model_id, + 'last_timestamp': last_timestamp }, retry_policy=retry_policy, start_to_close_timeout=timedelta(minutes=1), diff --git a/requirements.txt b/requirements.txt index 5604fc7..599917b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,5 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14 -git+ssh://git@github.com/Aignosi/sientia-mlops-library.git +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.17 +git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1 diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 6b61c81..28ec1d2 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -42,7 +42,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac side_effect=Exception("Test error")) input_data = { 'filters': { - 'EMPTY_DATA': {'POLICY': 'STOP'} + 'EMPTY_DATA': {'policy': 'STOP', 'config': {}} }, 'data': {'value': []}, 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'] @@ -55,7 +55,7 @@ async def test_input_gate_filter_exception(mock_input_filter_functions, gates_ac assert result == (None, 0, "") gates_activity.notification_handler.build_and_send_notification.assert_called_once_with( notification_id="INTPUT_GATE_ERROR__EMPTY_DATA", - message="Error in filter EMPTY_DATA:{'POLICY': 'STOP'}: \n Test error", + message="Error in filter EMPTY_DATA:{'policy': 'STOP', 'config': {}}: \n Test error", block="input_gate", level=NotificationLevel.ERROR, attachment_content=ANY @@ -84,7 +84,7 @@ async def test_input_gate_with_filter(gates_activity): # Arrange input_data = { 'filters': { - 'EMPTY_DATA': {'POLICY': 'STOP'} + 'EMPTY_DATA': {'policy': 'STOP', 'config': {}} }, 'data': {'value': []}, 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'] @@ -171,7 +171,7 @@ async def test_mlflow_response_gate_with_filter(gates_activity): # Arrange input_data = { 'filters': { - 'API_ERROR': {'POLICY': 'STOP'} + 'API_ERROR': {'policy': 'STOP'} }, 'data': { 'success': False, @@ -273,7 +273,7 @@ async def test_mlflow_content_gate_with_filter(gates_activity): # Arrange input_data = { 'filters': { - 'NAN_VALUES': {'POLICY': 'STOP'} + 'NAN_VALUES': {'policy': 'STOP', 'config': {}} }, 'data': {'value': [None, None, None]}, 'type': 'test', diff --git a/tests/laborious/utils/filters/test_conditional_filters.py b/tests/laborious/utils/filters/test_conditional_filters.py index edcbcd6..405bc9b 100644 --- a/tests/laborious/utils/filters/test_conditional_filters.py +++ b/tests/laborious/utils/filters/test_conditional_filters.py @@ -10,14 +10,14 @@ def test_filter_specific_variables_null_values(): assert filter_specific_variables_null_values( DataFrame( {'variable': ['variable1', 'variable2'], 'value': [1, 2]}), - config={'VARIABLES': ['variable2']}) is False + config={'variables': ['variable2']}) is False def test_filter_specific_variables_null_values_with_null_values(): assert filter_specific_variables_null_values( DataFrame( {'variable': ['variable1', 'variable2'], 'value': [1, None]}), - config={'VARIABLES': ['variable2']}) is True + config={'variables': ['variable2']}) is True def test_filter_empty_data(): diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 4318379..560565e 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -426,7 +426,8 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): { 'schema': schema, 'table_name': table_name, - 'model_id': model + 'model': model, + 'last_timestamp': last_timestamp, }, retry_policy=ANY, start_to_close_timeout=ANY diff --git a/values.yaml b/values.yaml index b164156..bd46925 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.0.2" + tag: "0.1.1" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-994-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas" + value: "SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux" - name: PYTHON_APP value: "laborious.worker.worker"