gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
28382c21905ebe5548d2536d405f58d48cc99442
sientia-dataops-orchestrato…/test.ipynb
vitor-aignosi 43206b578d SIENTIAPDE-1150 
fix: implement schedule normalization and timestamp management in MongoDB activities, enhance orchestration workflow with new formatting and normalization methods
2025-07-14 13:00:46 -03:00

651 lines
25 KiB
Plaintext
Raw Blame History

{
"cells": [
{
"cell_type": "code",
"execution_count": 4,
"id": "a287fa45",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Initializing Couchbase connection...\n",
"Awaiting Couchbase connection...\n",
"Couchbase connection ready\n"
]
}
],
"source": [
"from orchestrator.activities.couchbase import Couchbase\n",
"from unittest.mock import MagicMock\n",
"logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n",
"couchbase = Couchbase(\n",
" connection_string=\"couchbase://localhost\",\n",
" username=\"sientia\",\n",
" password=\"sientia\",\n",
" logger=logger,\n",
" notification_handler=MagicMock()\n",
")\n"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "d9a0f9c4",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Executing couchbase query: %s \n",
"SELECT\n",
" pipelines.*,\n",
" models as model\n",
"FROM\n",
" `pipelines`\n",
"JOIN\n",
" `models` ON KEYS pipelines.model_id;\n",
"\n",
"Fetched %d rows from couchbase 1\n",
"Rows: \n",
" %s [\n",
" {\n",
" \"filters\": [\n",
" {\n",
" \"filter_name\": \"OUT_OF_BOUNDS_FILTER\",\n",
" \"policy\": \"DISCARD\"\n",
" },\n",
" {\n",
" \"filter_name\": \"NULL_VALUES_FILTER\",\n",
" \"policy\": \"DISCARD\"\n",
" }\n",
" ],\n",
" \"frequency\": \"5s\",\n",
" \"max_retry_policy\": 1,\n",
" \"model\": {\n",
" \"name\": \"Demo Model-Demo2\"\n",
" },\n",
" \"model_id\": \"1\",\n",
" \"name\": \"scouter-opcua-pipeline\",\n",
" \"read_tags\": [\n",
" {\n",
" \"aggr_func\": \"avg\",\n",
" \"data_range\": [\n",
" -100,\n",
" 100\n",
" ],\n",
" \"tag_name\": \"Counter\"\n",
" }\n",
" ],\n",
" \"tag_retention_minutes\": 60,\n",
" \"workflow_type\": \"scouter\"\n",
" }\n",
"]\n",
"[{'filters': [{'filter_name': 'OUT_OF_BOUNDS_FILTER', 'policy': 'DISCARD'}, {'filter_name': 'NULL_VALUES_FILTER', 'policy': 'DISCARD'}], 'frequency': '5s', 'max_retry_policy': 1, 'model': {'name': 'Demo Model-Demo2'}, 'model_id': '1', 'name': 'scouter-opcua-pipeline', 'read_tags': [{'aggr_func': 'avg', 'data_range': [-100, 100], 'tag_name': 'Counter'}], 'tag_retention_minutes': 60, 'workflow_type': 'scouter'}]\n"
]
}
],
"source": [
"query = \"\"\"\n",
"SELECT\n",
" pipelines.*,\n",
" models as model\n",
"FROM\n",
" `pipelines`\n",
"JOIN\n",
" `models` ON KEYS pipelines.model_id;\n",
"\"\"\"\n",
"if __name__ == \"__main__\":\n",
"\n",
"\n",
" result = await couchbase.load_query_from_couchbase({\n",
" \"query\": query\n",
" })\n",
"\n",
" print(result)\n",
" "
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "7d01f160",
"metadata": {},
"outputs": [],
"source": [
"from temporalio import client\n",
"from orchestrator.activities.temporal_manager import TemporalManager\n",
"import os\n",
"from unittest.mock import MagicMock\n",
"\n",
"host = \"localhost:7233\"\n",
"logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n",
"\n",
"temporal_client = await client.Client.connect(\n",
" target_host=host,\n",
" namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')\n",
")\n",
"\n",
" "
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "bb750ae6",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<temporalio.client.ScheduleHandle at 0x7700871c6150>"
]
},
"execution_count": 4,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"import asyncio\n",
"from datetime import timedelta\n",
"from temporalio.client import (\n",
" Client,\n",
" Schedule,\n",
" ScheduleActionStartWorkflow,\n",
" ScheduleIntervalSpec,\n",
" ScheduleSpec,\n",
")\n",
"from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n",
"\n",
"# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\n",
"\n",
"\n",
"\n",
"await temporal_client.create_schedule(\n",
" \"orchestrator\",\n",
" Schedule(\n",
" action=ScheduleActionStartWorkflow(\n",
" 'orchestrator',\n",
" {\n",
" \"schedule_name\": \"orchestrator-test\",\n",
" \"pipelines_query\": {\n",
" \"collection\": \"pipelines\",\n",
" \"aggregation\": [\n",
" {\n",
" \"$lookup\": {\n",
" \"from\": \"models\",\n",
" \"localField\": \"model_id\",\n",
" \"foreignField\": \"id\",\n",
" \"as\": \"model_docs\"\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$addFields\": {\n",
" \"models\": {\n",
" \"$arrayElemAt\": [\n",
" \"$model_docs\",\n",
" 0\n",
" ]\n",
" }\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"models.active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$project\": {\n",
" \"model_docs\": 0\n",
" }\n",
" }\n",
" ]\n",
" },\n",
" \"opc_servers_query\": {\n",
" \"collection\": \"opc-servers\",\n",
" \"filters\": {\n",
"\n",
" }\n",
" }\n",
" },\n",
" id=\"orchestrator\",\n",
" task_queue=\"orchestrator-queue\",\n",
" execution_timeout=timedelta(minutes=600)\n",
" ),\n",
" spec=ScheduleSpec(\n",
" intervals=[ScheduleIntervalSpec(every=timedelta(minutes=60))]\n",
" )\n",
" )\n",
")\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b840347d",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Error deleting schedule scouter-load-test-1: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-2: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-3: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-4: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-5: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-6: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-7: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-8: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-9: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-10: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-11: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-12: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-13: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-14: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-15: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-16: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-17: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-18: workflow execution already completed\n",
"Error deleting schedule scouter-load-test-19: workflow execution already completed\n"
]
}
],
"source": [
"# Delete schedule by ID\n",
"schedule_id = \"scouter-load-test-num\"\n",
"\n",
"for i in range(0, 70):\n",
"\n",
" try:\n",
" handle = temporal_client.get_schedule_handle(\n",
" schedule_id.replace(\"num\", str(i)))\n",
" \n",
" await handle.delete()\n",
" except Exception as e:\n",
" print(f\"Error deleting schedule {schedule_id.replace('num', str(i))}: {e}\")\n",
"\n",
"# Delete schedule by ID\n",
"schedule_id = \"laborious-load-test-num\"\n",
"\n",
"for i in range(0, 70):\n",
"\n",
" try:\n",
" handle = temporal_client.get_schedule_handle(\n",
" schedule_id.replace(\"num\", str(i)))\n",
" \n",
" await handle.delete()\n",
" except Exception as e:\n",
" print(f\"Error deleting schedule {schedule_id.replace('num', str(i))}: {e}\")\n"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "374b5b0e",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"scouter-load-test-num\n"
]
}
],
"source": [
"schedule_id = \"scouter-load-test-num\"\n",
"schedule_id.replace(\"-\", \"_\")\n",
"\n",
"print(schedule_id)"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "1bd82225",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Getting orchestrated schedules...\n"
]
}
],
"source": [
"schedules = await manager.load_schedule()"
]
},
{
"cell_type": "code",
"execution_count": 24,
"id": "1948670e",
"metadata": {},
"outputs": [],
"source": [
"from google.protobuf.json_format import MessageToDict\n",
"import base64\n",
"import json\n",
"for arg in schedules.schedule.action.args:\n",
" data = MessageToDict(arg)\n",
" data = base64.b64decode(data['data']).decode('utf-8')"
]
},
{
"cell_type": "code",
"execution_count": 12,
"id": "a4c777dd",
"metadata": {},
"outputs": [],
"source": [
"from google.protobuf.json_format import MessageToDict\n",
"import base64\n",
"import json\n",
"\n",
"schedules_config = {}\n",
"async for schedule in await temporal_client.list_schedules():\n",
" id = schedule.id\n",
"\n",
" handle = temporal_client.get_schedule_handle(id)\n",
"\n",
" desc = await handle.describe()\n",
"\n",
" for arg in desc.schedule.action.args:\n",
" data = MessageToDict(arg)['data']\n",
" data = base64.b64decode(data).decode('utf-8')\n",
"\n",
" frequency = desc.schedule.spec.intervals[0].every.seconds\n",
"\n",
" schedules_config[id] = {\n",
" 'frequency': frequency,\n",
" 'data': json.loads(data),\n",
" 'handle': handle\n",
" }\n",
" \n",
" \n",
" \n",
" "
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "f81b3728",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'id': 'orchestrator', 'schedule': ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='orchestrator'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=3600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), 'info': ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 9, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 9, 0, 0, 165083, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T09:00:00Z', first_execution_run_id='01980829-9700-771a-9c9d-2450eecc9af4')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 10, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 10, 0, 0, 125309, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T10:00:00Z', first_execution_run_id='01980860-8578-759d-97fa-dbdfc784f88a')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 11, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 11, 0, 0, 124356, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T11:00:00Z', first_execution_run_id='01980897-73f7-7e83-a7a6-20eaac38e7d4')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 12, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 12, 0, 0, 167329, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T12:00:00Z', first_execution_run_id='019808ce-62a2-75d6-b11d-c5387cfa258f')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 7, 14, 13, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 7, 14, 13, 0, 0, 167575, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='orchestrator-2025-07-14T13:00:00Z', first_execution_run_id='01980905-5122-7807-90ae-3782d0332e79'))], next_action_times=[datetime.datetime(2025, 7, 14, 14, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 15, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 16, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 17, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 7, 14, 18, 0, tzinfo=datetime.timezone.utc)]), 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'search_attributes': {}, 'data_converter': DataConverter(payload_converter_class=<class 'temporalio.converter.DefaultPayloadConverter'>, payload_codec=None, failure_converter_class=<class 'temporalio.converter.DefaultFailureConverter'>, payload_converter=<temporalio.converter.DefaultPayloadConverter object at 0x787d51390750>, failure_converter=<temporalio.converter.DefaultFailureConverter object at 0x787d5277b590>), 'raw_entry': schedule_id: \"orchestrator\"\n",
"info {\n",
" spec {\n",
" interval {\n",
" interval {\n",
" seconds: 3600\n",
" }\n",
" }\n",
" }\n",
" workflow_type {\n",
" name: \"orchestrator\"\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752483600\n",
" }\n",
" actual_time {\n",
" seconds: 1752483600\n",
" nanos: 165083498\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T09:00:00Z\"\n",
" run_id: \"01980829-9700-771a-9c9d-2450eecc9af4\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752487200\n",
" }\n",
" actual_time {\n",
" seconds: 1752487200\n",
" nanos: 125309119\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T10:00:00Z\"\n",
" run_id: \"01980860-8578-759d-97fa-dbdfc784f88a\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752490800\n",
" }\n",
" actual_time {\n",
" seconds: 1752490800\n",
" nanos: 124356906\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T11:00:00Z\"\n",
" run_id: \"01980897-73f7-7e83-a7a6-20eaac38e7d4\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752494400\n",
" }\n",
" actual_time {\n",
" seconds: 1752494400\n",
" nanos: 167329351\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T12:00:00Z\"\n",
" run_id: \"019808ce-62a2-75d6-b11d-c5387cfa258f\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n",
" }\n",
" recent_actions {\n",
" schedule_time {\n",
" seconds: 1752498000\n",
" }\n",
" actual_time {\n",
" seconds: 1752498000\n",
" nanos: 167575982\n",
" }\n",
" start_workflow_result {\n",
" workflow_id: \"orchestrator-2025-07-14T13:00:00Z\"\n",
" run_id: \"01980905-5122-7807-90ae-3782d0332e79\"\n",
" }\n",
" start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n",
" }\n",
" future_action_times {\n",
" seconds: 1752501600\n",
" }\n",
" future_action_times {\n",
" seconds: 1752505200\n",
" }\n",
" future_action_times {\n",
" seconds: 1752508800\n",
" }\n",
" future_action_times {\n",
" seconds: 1752512400\n",
" }\n",
" future_action_times {\n",
" seconds: 1752516000\n",
" }\n",
"}\n",
"}\n",
"{'_client': <temporalio.client.Client object at 0x787d53dd3490>, 'id': 'orchestrator'}\n"
]
}
],
"source": [
"async for schedule in await temporal_client.list_schedules():\n",
" print(vars(schedule))\n",
" id = schedule.id\n",
"\n",
" handle = temporal_client.get_schedule_handle(id)\n",
"\n",
" print(vars(handle))"
]
},
{
"cell_type": "code",
"execution_count": 13,
"id": "988d1718",
"metadata": {},
"outputs": [],
"source": [
"schedules_config = {\n",
" \"scouter-opcua-orchestrated-pipeline\": schedules_config['scouter-opcua-orchestrated-pipeline'],\n",
"}"
]
},
{
"cell_type": "code",
"execution_count": 14,
"id": "c12f5e75",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"scouter-opcua-orchestrated-pipeline {'frequency': 5, 'data': {'filters': {'NULL_VALUES_FILTER': {'policy': 'DISCARD'}, 'OUT_OF_BOUNDS_FILTER': {'policy': 'DISCARD'}}, 'frequency': '5s', 'max_retry_policy': 1, 'model_id': '1', 'model_name': 'Demo Model-Demo2', 'model_tags': {'Counter': {'aggr_func': 'avg', 'data_range': [-100, 100]}}, 'retention_time': 3600, 'schedule_name': 'scouter-opcua-orchestrated-pipeline', 'schema': 'sientia_data', 'table_name': 'laborious_data', 'topic': 'raw_scouter-opcua-orchestrated-pipeline', 'trigger_laborious': False, 'workflow_type': 'scouter'}, 'handle': <temporalio.client.ScheduleHandle object at 0x7dfab3b2ab50>}\n",
"scouter-opcua-orchestrated-pipeline\n",
"{'workflow': 'scouter', 'args': [metadata {\n",
" key: \"encoding\"\n",
" value: \"json/plain\"\n",
"}\n",
"data: \"{\\\"filters\\\":{\\\"NULL_VALUES_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"},\\\"OUT_OF_BOUNDS_FILTER\\\":{\\\"policy\\\":\\\"DISCARD\\\"}},\\\"frequency\\\":\\\"5s\\\",\\\"max_retry_policy\\\":1,\\\"model_id\\\":\\\"1\\\",\\\"model_name\\\":\\\"Demo Model-Demo2\\\",\\\"model_tags\\\":{\\\"Counter\\\":{\\\"aggr_func\\\":\\\"avg\\\",\\\"data_range\\\":[-100,100]}},\\\"retention_time\\\":3600,\\\"schedule_name\\\":\\\"scouter-opcua-orchestrated-pipeline\\\",\\\"schema\\\":\\\"sientia_data\\\",\\\"table_name\\\":\\\"laborious_data\\\",\\\"topic\\\":\\\"raw_scouter-opcua-orchestrated-pipeline\\\",\\\"trigger_laborious\\\":false,\\\"workflow_type\\\":\\\"scouter\\\"}\"\n",
"], 'id': 'scouter-opcua-orchestrated-pipeline', 'task_queue': 'scouter-queue', 'execution_timeout': datetime.timedelta(seconds=120), 'run_timeout': None, 'task_timeout': None, 'retry_policy': None, 'memo': None, 'typed_search_attributes': TypedSearchAttributes(search_attributes=[]), 'headers': None, 'untyped_search_attributes': {}, 'static_summary': None, 'static_details': None, 'priority': Priority(priority_key=None)}\n"
]
}
],
"source": [
"from temporalio.client import Client, ScheduleUpdateInput, ScheduleUpdate, ScheduleSpec\n",
"\n",
"\n",
"for id, schedule in schedules_config.items():\n",
" print(id, schedule)\n",
" \n",
" if schedule['frequency'] != 600:\n",
" print(id)\n",
"\n",
" handle = schedule['handle']\n",
" \n",
" async def update_schedule(input: ScheduleUpdateInput) -> ScheduleUpdate:\n",
" schedule_action = input.description.schedule.action\n",
"\n",
" print(schedule_action.__dict__)\n",
" \n",
" if hasattr(schedule_action, 'args'):\n",
" schedule_action.args = [{}]\n",
" \n",
" # Atualiza o intervalo de execução\n",
" input.description.schedule.spec.intervals = [\n",
" ScheduleIntervalSpec(every=timedelta(minutes=10))\n",
" ]\n",
" \n",
" return ScheduleUpdate(schedule=input.description.schedule)\n",
"\n",
" await handle.update(update_schedule)\n",
" \n"
]
},
{
"cell_type": "code",
"execution_count": 1,
"id": "87eb10c8",
"metadata": {},
"outputs": [],
"source": [
"from redis import Redis\n",
"\n",
"redis = Redis(host='localhost', port=6379)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "fe4ad9dc",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Chave: slot:opc_tags:2, Valor: {\n",
"\"slot2\": \"value\"\n",
"}\n",
"Chave: slot:opc_tags:1, Valor: {\n",
"\"slot1\": \"value\"\n",
"}\n"
]
}
],
"source": [
"matching_keys = redis.keys(\"slot:opc_tags:*\")\n",
"\n",
"if matching_keys:\n",
" decoded_keys = [key.decode('utf-8') for key in matching_keys]\n",
" values = redis.mget(decoded_keys)\n",
"\n",
" items = {}\n",
" for i, key in enumerate(decoded_keys):\n",
" value = values[i]\n",
" if value is not None:\n",
" try:\n",
" items[key] = value.decode('utf-8')\n",
" except (UnicodeDecodeError, AttributeError):\n",
" items[key] = value\n",
" else:\n",
" items[key] = None\n",
"\n",
" for key, value in items.items():\n",
" print(f\"Chave: {key}, Valor: {value}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "87a9dc89",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.13"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
Reference in New Issue View Git Blame Copy Permalink