gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #28 from Aignosi/fix/SIENTIAPDE-1314-ajustes-nas-camadas-de-monitoramento-do-sientia

Browse Source 
Refactor Quality Gate Workflow for Enhanced Configuration and Streamlined Analysis
This commit is contained in:
Matheus Demoner
2025-10-27 14:26:50 -03:00
committed by GitHub
parent ec3ab3c936 1e6616cfc2
commit f099366bed
8 changed files with 130 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
encrypt.py Normal file
View File

@@ -0,0 +1,113 @@
import os
import argparse
from pathspec import PathSpec
import yaml # type: ignore
from typing import Any
'''
Usage:
python .\encrypt.py path_to_dir output_file --ignore ignore_file --chunk-size 100000
'''
def load_ignore_patterns(ignore_file, include_library):
# Ensure the .gitignore file exists
if not os.path.exists(ignore_file):
raise FileNotFoundError(f"Ignore file not found at {ignore_file}")
# Load and parse the .gitignore patterns
with open(ignore_file, 'r') as file:
patterns = file.readlines()
if not include_library:
patterns.append('**/deploy/library/')
spec = PathSpec.from_lines('gitwildmatch', patterns)
return spec
def is_ignored(file_path, spec):
"""Check if a file should be ignored based on the ignore patterns."""
return spec.match_file(file_path) if spec else False
def encode_file_tree_to_yaml(directory, ignore_file, include_library):
"""Encode the file tree into a single YAML file."""
ignore_patterns = load_ignore_patterns(
ignore_file, include_library) if ignore_file else None
file_tree: dict[str, Any] = {}
for root, dirs, files in os.walk(directory):
# Skip ignored directories
dirs[:] = [d for d in dirs if not is_ignored(
os.path.join(root, d), ignore_patterns)]
for file in files:
file_path = os.path.join(root, file)
# Skip ignored files
if is_ignored(file_path, ignore_patterns):
continue
# Read file content
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
except Exception as e:
print(f"Error reading file {file_path}: {e}")
raise
# Create nested dictionary structure
path_parts = os.path.relpath(file_path, directory).split(os.sep)
current_level = file_tree
# all except the last part (the file name)
for part in path_parts[:-1]:
current_level = current_level.setdefault(part, {})
# Add the file and its content
current_level[path_parts[-1]] = content
return yaml.dump(file_tree, default_flow_style=False)
def chunk_and_write_file_tree_to_yaml(yaml_content, output_file, chunk_size=None):
"""Chunk the YAML content and write it to the output file."""
chunks = [yaml_content] if chunk_size is None else [
yaml_content[i:i + chunk_size] for i in range(0, len(yaml_content), chunk_size)]
for i, chunk in enumerate(chunks):
chunk_file = f"{output_file}_{i}.yaml"
# Write the file tree to the output YAML file
with open(chunk_file, 'w', encoding='utf-8') as yaml_file:
yaml_file.write(chunk)
def main():
parser = argparse.ArgumentParser(
description="Encrypts file tree to yaml file")
parser.add_argument("input_directory", help="Directory to encode")
parser.add_argument("output_yaml_file", help="Output YAML file")
parser.add_argument("--ignore", default=None,
help="Path to the ignore file")
parser.add_argument("--chunk-size", type=int, default=None,
help="Chunk size for the output YAML file")
parser.add_argument("--library", type=bool, default=False,
help="Incude the library in the output YAML file")
# Parse arguments
args = parser.parse_args()
# Example usage
directory_to_encode = args.input_directory
ignore_file_path = args.ignore
output_yaml_file = args.output_yaml_file
include_library = args.library
content = encode_file_tree_to_yaml(
directory_to_encode, ignore_file_path, include_library)
chunk_and_write_file_tree_to_yaml(
content, output_yaml_file, args.chunk_size)
if __name__ == "__main__":
main()

6
orchestrator/workflows/alerts.py
View File

@@ -67,7 +67,9 @@ class Alerts:
# Call subworkflow "load_notification_package" passing the static filters
# (level = "ERROR" and timestamp > last timestamp)
package = await workflow.execute_child_workflow('load_notification_package', input_data)
package = await workflow.execute_child_workflow(
'subworkflow.load_notification_package', input_data
)
if not package['notification_package'] or not package['sending_configs']:
return
@@ -92,7 +94,7 @@ class Alerts:
# Call subworkflow "process_notifications" passing the notification package
log_report = await workflow.execute_child_workflow(
'process_notifications',
'subworkflow.process_notifications',
{
'metadata': metadata,
'mail_type': mail_type,

6
orchestrator/workflows/reports.py
View File

@@ -59,7 +59,9 @@ class Reports:
# Call subworkflow "load_notification_package" passing the static filters
# (timestamp > last timestamp)
package = await workflow.execute_child_workflow('load_notification_package', input_data)
package = await workflow.execute_child_workflow(
'subworkflow.load_notification_package', input_data
)
if not package['notification_package'] or not package['sending_configs']:
return
@@ -83,7 +85,7 @@ class Reports:
# Call subworkflow "process_notifications" passing the notification package
await workflow.execute_child_workflow(
'process_notifications',
'subworkflow.process_notifications',
{
'metadata': metadata,
'mail_type': mail_type,

2
orchestrator/workflows/subworkflows/load_notification_package.py
View File

@@ -9,7 +9,7 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
@workflow.defn(name='load_notification_package')
@workflow.defn(name='subworkflow.load_notification_package')
class LoadNotificationPackage:
"""
Subworkflow for loading notification data and configuration.

2
orchestrator/workflows/subworkflows/process_notifications.py
View File

@@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
@workflow.defn(name='process_notifications')
@workflow.defn(name='subworkflow.process_notifications')
class ProcessNotifications:
"""
Subworkflow for processing and sending notification emails.

6
tests/orchestrator/workflows/test_alerts.py
View File

@@ -31,7 +31,7 @@ async def test_run_full_flow(workflow_mock, alerts):
workflow_mock.execute_child_workflow.assert_has_calls(
[
call(
'load_notification_package',
'subworkflow.load_notification_package',
{**input_data, 'metadata': metadata, 'base_data_filter': {'level': 'ERROR'}},
)
]
@@ -40,7 +40,7 @@ async def test_run_full_flow(workflow_mock, alerts):
workflow_mock.execute_child_workflow.assert_has_calls(
[
call(
'process_notifications',
'subworkflow.process_notifications',
{
'metadata': metadata,
'mail_type': 'Alerts',
@@ -104,7 +104,7 @@ async def test_run_no_data(workflow_mock, alerts):
workflow_mock.execute_child_workflow.assert_has_calls(
[
call(
'load_notification_package',
'subworkflow.load_notification_package',
{**input_data, 'metadata': metadata, 'base_data_filter': {'level': 'ERROR'}},
)
]

6
tests/orchestrator/workflows/test_reports.py
View File

@@ -31,7 +31,7 @@ async def test_run_full_flow(workflow_mock, reports):
workflow_mock.execute_child_workflow.assert_has_calls(
[
call(
'load_notification_package',
'subworkflow.load_notification_package',
{**input_data, 'metadata': metadata, 'base_data_filter': {}},
)
]
@@ -40,7 +40,7 @@ async def test_run_full_flow(workflow_mock, reports):
workflow_mock.execute_child_workflow.assert_has_calls(
[
call(
'process_notifications',
'subworkflow.process_notifications',
{
'metadata': metadata,
'mail_type': 'Reports',
@@ -88,7 +88,7 @@ async def test_run_no_data(workflow_mock, reports):
workflow_mock.execute_child_workflow.assert_has_calls(
[
call(
'load_notification_package',
'subworkflow.load_notification_package',
{**input_data, 'metadata': metadata, 'base_data_filter': {}},
)
]

2
values.yaml
View File

@@ -151,7 +151,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados"
value: "fix/SIENTIAPDE-1314-ajustes-nas-camadas-de-monitoramento-do-sientia"
- name: PYTHON_APP
value: "orchestrator.worker.worker"