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

SIENTIAPDE-1974

Browse Source 
SIENTIAPDE-1974 Extract PI tag resolution and name Mongo stage constants

- Move PI Web API read/output tag resolution out of the schedule loop
  into _resolve_pi_pipeline, so the loop only handles the held case.
- Replace repeated '$match'/'$project' literals in the default
  connector queries with MATCH/PROJECT constants.
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-28 15:52:35 -03:00
parent 7e6f1a8782
commit 16e76f11ea
2 changed files with 57 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
orchestrator/activities/formatters.py
View File

@@ -99,6 +99,46 @@ def _merge_issues(
issues.setdefault(kind, set()).update(f'{schedule_name}/{value}' for value in values)
def _resolve_pi_pipeline(
pipeline: dict[str, Any],
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
issues: dict[str, set[str]],
) -> dict[str, Any] | None:
"""
The pipeline with its PI Web API tags resolved through the tag database.
pi_web_api_scouter read tags get their WebIDs (and the pipeline its connector);
predictions_batch output tags get {tag_name: WebID}. Other pipelines come back as
they are. What cannot be resolved is added to issues. None means the pipeline must
be held: its read tags span several connectors.
"""
workflow_type = pipeline['workflow_type']
schedule_name = pipeline['schedule_name']
if workflow_type == 'pi_web_api_scouter' and isinstance(pipeline.get('read_tags'), dict):
read_tags, connector, read_issues = resolve_pi_read_tags(
pipeline['read_tags'], pi_tags, pi_connectors
)
_merge_issues(issues, read_issues, schedule_name)
if read_issues['mixed']:
return None
pipeline = {**pipeline, 'read_tags': read_tags}
if connector is not None:
pipeline['pi_web_api_connector'] = connector
if workflow_type in PI_OUTPUT_WORKFLOW_TYPES and isinstance(
pipeline.get('pi_web_api_output_config'), dict
):
output_config, output_issues = resolve_pi_output_tags(
pipeline['pi_web_api_output_config'], pi_tags, pi_connectors
)
_merge_issues(issues, output_issues, schedule_name)
pipeline = {**pipeline, 'pi_web_api_output_config': output_config}
return pipeline
class ScheduleType(TypedDict):
"""
Type definition for schedule type configuration entries.
@@ -292,28 +332,11 @@ class Formatters(SientiaMonitoring):
'read_tags': resolve_tag_names(pipeline['read_tags'], tags_by_id),
}
if workflow_type == 'pi_web_api_scouter' and isinstance(
pipeline.get('read_tags'), dict
):
read_tags, connector, issues = resolve_pi_read_tags(
pipeline['read_tags'], pi_tags, pi_connectors
)
_merge_issues(pi_issues, issues, schedule_name)
if issues['mixed']:
schedule_config[namespace][schedule_name] = {'held': True}
continue
pipeline = {**pipeline, 'read_tags': read_tags}
if connector is not None:
pipeline['pi_web_api_connector'] = connector
if workflow_type in PI_OUTPUT_WORKFLOW_TYPES and isinstance(
pipeline.get('pi_web_api_output_config'), dict
):
output_config, issues = resolve_pi_output_tags(
pipeline['pi_web_api_output_config'], pi_tags, pi_connectors
)
_merge_issues(pi_issues, issues, schedule_name)
pipeline = {**pipeline, 'pi_web_api_output_config': output_config}
resolved = _resolve_pi_pipeline(pipeline, pi_tags, pi_connectors, pi_issues)
if resolved is None:
schedule_config[namespace][schedule_name] = {'held': True}
continue
pipeline = resolved
schedule_config[namespace][schedule_name] = {
**function(pipeline),

20
orchestrator/workflows/orchestrator.py
View File

@@ -14,6 +14,10 @@ with workflow.unsafe.imports_passed_through():
referenced_tag_ids,
)
# MongoDB aggregation stages
MATCH = '$match'
PROJECT = '$project'
# Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow
# input may override these aggregations (connectors_query / connector_tags_query); the
# workflow puts a tag_id $match in front of connector_tags_query. Only the fields the
@@ -22,9 +26,9 @@ with workflow.unsafe.imports_passed_through():
DEFAULT_CONNECTORS_QUERY = {
'collection': 'connectors',
'aggregation': [
{'$match': {'type': 'opc_ua', 'status': 'Active'}},
{MATCH: {'type': 'opc_ua', 'status': 'Active'}},
{
'$project': {
PROJECT: {
'_id': 0,
'connector_id': 1,
'name': 1,
@@ -39,8 +43,8 @@ DEFAULT_CONNECTORS_QUERY = {
DEFAULT_CONNECTOR_TAGS_QUERY = {
'collection': 'connector_tags',
'aggregation': [
{'$match': {'category': 'Read', 'status': 'Active'}},
{'$project': {'_id': 0, 'tag_id': 1, 'connector_id': 1, 'address': 1, 'tag_name': 1}},
{MATCH: {'category': 'Read', 'status': 'Active'}},
{PROJECT: {'_id': 0, 'tag_id': 1, 'connector_id': 1, 'address': 1, 'tag_name': 1}},
],
}
# Runs started before the tag database loads existed replay without them
@@ -53,16 +57,16 @@ TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database'
DEFAULT_PI_CONNECTORS_QUERY = {
'collection': 'connectors',
'aggregation': [
{'$match': {'type': 'pi_web_api', 'status': 'Active'}},
{'$project': {'_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, 'pi_web_api': 1}},
{MATCH: {'type': 'pi_web_api', 'status': 'Active'}},
{PROJECT: {'_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, 'pi_web_api': 1}},
],
}
DEFAULT_PI_CONNECTOR_TAGS_QUERY = {
'collection': 'connector_tags',
'aggregation': [
{'$match': {'status': 'Active'}},
{MATCH: {'status': 'Active'}},
{
'$project': {
PROJECT: {
'_id': 0,
'tag_id': 1,
'connector_id': 1,