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

Merge pull request #3 from Aignosi/SIENTIAPDE-1094-atualizar-versao-da-lib

Browse Source 
Sientiapde 1094 atualizar versao da lib
This commit is contained in:
Bruno Domingues
2025-06-09 13:33:51 -03:00
committed by GitHub
parent deafb335a1 b8beb0e151
commit ab0c21f5fb
21 changed files with 51 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.env
View File

@@ -6,13 +6,13 @@ POSTGRES_DB=sientia
POSTGRES_MIN_CONNECTIONS=5
POSTGRES_MAX_CONNECTIONS=20
KAFKA_BOOTSTRAP_SERVERS=kafka:29092
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_POLLING_TIME=1000
REDIS_HOST=redis
REDIS_HOST=localhost
REDIS_PORT=6379
TEMPORAL_HOST=host.docker.internal:7233
TEMPORAL_HOST=localhost:7233
TEMPORAL_NAMESPACE=default
LOG_LEVEL=INFO

2
requirements.txt
View File

@@ -3,5 +3,5 @@ psycopg2-binary
sqlalchemy
asyncua
redis
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git

4
scouter/activities/activities.py
View File

@@ -1,12 +1,12 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from scouter.activities.postgres import Postgres
from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.redis import Redis
from scouter.activities.kafka import Kafka
from scouter.activities.gates import Gates
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from typing import Any

26
scouter/activities/base.py
View File

@@ -1,26 +0,0 @@
from typing import Any
from logging import Logger
from temporalio import activity
from sientia_do.notifications.handlers import NotificationHandler
class BaseActivity:
def __init__(self, logger: Logger, notification_handler: NotificationHandler):
self.logger = logger
self.notification_handler = notification_handler
@activity.defn(name="prepare_activity")
async def prepare_activity(self, input_data: dict[str, Any]):
"""
Prepare the activity for the notification handler.
Args:
workflow_name (str): The name of the workflow.
schedule_name (str): The name of the schedule.
model_name (str): The name of the model.
model_id (str): The id of the model.
"""
self.notification_handler.base_notification.pipeline_name = input_data['workflow_name']
self.notification_handler.base_notification.schedule_name = input_data['schedule_name']
self.notification_handler.base_notification.model_name = input_data['model_name']
self.notification_handler.base_notification.model_id = input_data['model_id']

2
scouter/activities/faker.py
View File

@@ -7,7 +7,7 @@ from kafka import KafkaProducer
from temporalio import activity
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.base import BaseActivity
from sientia_do.temporal.activities.base import BaseActivity
class Faker(BaseActivity):

2
scouter/activities/gates.py
View File

@@ -2,7 +2,7 @@ from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.base import BaseActivity
from sientia_do.temporal.activities.base import BaseActivity
from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
from typing import Any
import traceback

2
scouter/activities/kafka.py
View File

@@ -3,7 +3,7 @@ from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.base import BaseActivity
from sientia_do.temporal.activities.base import BaseActivity
from typing import Any
from kafka import KafkaConsumer
from pandas import DataFrame

85
scouter/activities/postgres.py
View File

@@ -1,85 +0,0 @@
import traceback
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool
from pandas import DataFrame
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.base import BaseActivity
from typing import Any
class Postgres(BaseActivity):
def __init__(self, host: str, port: int,
user: str, password: str, dbname: str,
min_connections: int, max_connections: int,
logger: Logger, notification_handler: NotificationHandler):
self.host = host
self.port = port
self.user = user
self.password = password
self.dbname = dbname
# Create SQLAlchemy engine with connection pooling
self.engine = create_engine(
f'postgresql://{user}:{password}@{host}:{port}/{dbname}',
poolclass=QueuePool,
pool_size=min_connections,
max_overflow=max_connections - min_connections,
pool_pre_ping=True
)
self.session_factory = sessionmaker(bind=self.engine)
BaseActivity.__init__(self, logger, notification_handler)
def close(self):
self.engine.dispose()
def __del__(self):
self.close()
@activity.defn(name="export_data_to_postgres")
async def export_data_to_postgres(self, input_data: dict[str, Any]):
"""
Exports data to a postgres table.
Args:
input_data (dict[str, Any]): The data to export. Contains:
schema (str): The schema of the table.
table_name (str): The name of the table.
data (DataFrame): The data to export.
"""
self.logger.debug(
f"Exporting data to postgres: {input_data['data']}")
schema = input_data["schema"]
table_name = input_data["table_name"]
data = DataFrame(input_data["data"])
with self.session_factory() as session:
try:
data.to_sql(table_name, self.engine, schema=schema,
if_exists="append", index=False)
session.commit()
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES",
message=f"Error exporting data to postgres: {e}",
block="export_data_to_postgres",
level=NotificationLevel.ERROR,
attachment_content=trace
)
self.logger.error(trace)
else:
self.logger.debug("Data exported to postgres")
finally:
session.close()

28
scouter/activities/redis.py
View File

@@ -3,39 +3,19 @@ from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
from logging import Logger
from sientia_do.notifications.handlers import NotificationHandler
from scouter.activities.base import BaseActivity
import redis
import json
from sientia_do.temporal.activities.redis_base import Redis as RedisBase
from typing import Any
from pandas import DataFrame
from datetime import datetime
class Redis(BaseActivity):
class Redis(RedisBase):
def __init__(self, host: str, port: int,
username: str, password: str,
logger: Logger, notification_handler: NotificationHandler):
self.host = host
self.port = port
self.username = username
self.password = password
self.redis_client = redis.Redis(
host=self.host,
port=self.port,
decode_responses=True,
username=self.username,
password=self.password
)
BaseActivity.__init__(self, logger, notification_handler)
def get(self, key: str):
history = self.redis_client.get(key)
return json.loads(history) if history else None
def set(self, key: str, data: dict, ttl=600):
self.redis_client.set(key, json.dumps(data), ex=ttl)
RedisBase.__init__(self, host, port, username,
password, logger, notification_handler)
@activity.defn(name="group_and_hold_data")
async def group_and_hold_data(self, input_data: dict[str, Any]):

22
scouter/utils/logger.py
View File

@@ -1,22 +0,0 @@
from os import getenv
import logging
import sys
def get_logger(name: str):
log_level = getenv('LOG_LEVEL', 'INFO').upper()
logger = logging.getLogger(name)
logger.setLevel(log_level)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setLevel(log_level)
stream_handler.setFormatter(
logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
)
logger.addHandler(stream_handler)
return logger

9
scouter/utils/policies.py
View File

@@ -1,9 +0,0 @@
from temporalio.common import RetryPolicy
from datetime import timedelta
retry_policy = RetryPolicy(
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
maximum_interval=timedelta(minutes=1),
maximum_attempts=1
)

8
scouter/worker/worker.py
View File

@@ -4,13 +4,13 @@ from temporalio.worker import Worker
with workflow.unsafe.imports_passed_through():
import os
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.temporal.utils.logger import get_logger
from scouter.activities.activities import Activities
from scouter.workflow.scouter import Scouter
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
from scouter.workflow.fake_data import FakeData
from scouter.activities.faker import Faker
import asyncio
from scouter.utils.logger import get_logger
from scouter.utils.connectors_config import (
build_postgres_config,
build_kafka_config,
@@ -30,11 +30,7 @@ async def main():
notification_handler = NotificationHandler(
servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'),
logger=logger,
project_name=os.getenv('PROJECT_NAME', 'scouter'),
pipeline_name='-',
trigger_name='-',
model_name='-',
model='-'
project_name=os.getenv('PROJECT_NAME', 'scouter')
)
logger.info('Starting Activities...')

2
scouter/workflow/fake_data.py
View File

@@ -4,7 +4,7 @@ with workflow.unsafe.imports_passed_through():
from scouter.activities.faker import Faker
from datetime import timedelta
from typing import Dict, Any
from scouter.utils.policies import retry_policy
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="fake_data")

2
scouter/workflow/scouter.py
View File

@@ -4,7 +4,7 @@ with workflow.unsafe.imports_passed_through():
from scouter.activities.activities import Activities
from typing import Any
from datetime import timedelta
from scouter.utils.policies import retry_policy
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="scouter")

2
scouter/workflow/sub_workflows/core_scouter.py
View File

@@ -4,7 +4,7 @@ with workflow.unsafe.imports_passed_through():
from scouter.activities.activities import Activities
from typing import Any
from datetime import timedelta
from scouter.utils.policies import retry_policy
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="core_scouter")

8
tests/activities/test_activities.py
View File

@@ -1,10 +1,10 @@
from unittest.mock import patch, MagicMock, ANY
from pytest import mark
from sientia_do.temporal.activities.postgres import Postgres
from scouter.activities.activities import Activities
from scouter.activities.postgres import Postgres
from scouter.activities.redis import Redis
from scouter.activities.kafka import Kafka
from scouter.activities.gates import Gates
from pytest import mark
@patch('scouter.activities.activities.Postgres.__init__')
@@ -141,9 +141,9 @@ async def test_prepare_activity(_mock_kafka_init,
await activities.prepare_activity(input_data)
assert activities.notification_handler.base_notification.pipeline_name == input_data[
assert activities.notification_handler.base_notification.pipeline == input_data[
'workflow_name']
assert activities.notification_handler.base_notification.schedule_name == input_data[
assert activities.notification_handler.base_notification.trigger == input_data[
'schedule_name']
assert activities.notification_handler.base_notification.model_name == input_data[
'model_name']

37
tests/activities/test_base.py
View File

@@ -1,37 +0,0 @@
from unittest.mock import MagicMock
from pytest import fixture, mark
from sientia_do.notifications.models import Notification
from scouter.activities.base import BaseActivity
@fixture
def base_activity():
return BaseActivity(
logger=MagicMock(),
notification_handler=MagicMock(),
)
@mark.asyncio
async def test_prepare_activity(base_activity):
base_activity.notification_handler.base_notification = Notification(
project="project",
pipeline="pipeline",
trigger="-",
model_name="-",
model_id="-",
)
await base_activity.prepare_activity(
{
'workflow_name': 'test_workflow',
'schedule_name': 'test_schedule',
'model_name': 'test_model',
'model_id': 'test_model_id',
}
)
assert base_activity.notification_handler.base_notification.schedule_name == "test_schedule"
assert base_activity.notification_handler.base_notification.model_name == "test_model"
assert base_activity.notification_handler.base_notification.model_id == "test_model_id"
assert base_activity.notification_handler.base_notification.pipeline_name == "test_workflow"

90
tests/activities/test_postgres.py
View File

@@ -1,90 +0,0 @@
from unittest.mock import ANY, MagicMock, patch
from pytest import fixture
from pytest import mark
from sientia_do.notifications.models import NotificationLevel
from scouter.activities.postgres import Postgres
@fixture
@patch("scouter.activities.postgres.create_engine")
@patch("scouter.activities.postgres.sessionmaker")
def postgres_client(mock_sessionmaker, mock_engine):
# Create a mock session
mock_session = MagicMock()
mock_session.commit = MagicMock()
mock_session.close = MagicMock()
# Configure the session to work with context management
mock_session.__enter__ = MagicMock(return_value=mock_session)
mock_session.__exit__ = MagicMock(return_value=None)
# Configure the sessionmaker to return our mock session
mock_sessionmaker.return_value = mock_session
# Configure the engine to return our mock sessionmaker
mock_engine.return_value = MagicMock()
mock_engine.return_value.dispose = MagicMock()
# Create the Postgres client
client = Postgres(
host="localhost",
port=5432,
user="postgres",
password="postgres",
dbname="postgres",
min_connections=1,
max_connections=10,
logger=MagicMock(),
notification_handler=MagicMock(),
)
# Set up the session factory
client.session_factory = mock_sessionmaker
return client
@mark.asyncio
@patch("scouter.activities.postgres.DataFrame")
async def test_export_data_to_postgres_success(mock_dataframe, postgres_client):
data = {"schema": "test", "table_name": "test",
"data": {"a": [1, 2, 3], "b": [4, 5, 6]}}
await postgres_client.export_data_to_postgres(data)
# Verify notification handler wasn't called
postgres_client.notification_handler.build_and_send_notification.assert_not_called()
# Verify session handling
mock_dataframe.assert_called_once_with(data["data"])
mock_dataframe.return_value.to_sql.assert_called_once_with(
data["table_name"],
postgres_client.engine,
schema=data["schema"],
if_exists="append",
index=False
)
postgres_client.session_factory.return_value.commit.assert_called_once()
postgres_client.session_factory.return_value.close.assert_called_once()
@mark.asyncio
@patch("scouter.activities.postgres.DataFrame", return_value=MagicMock(
to_sql=MagicMock(side_effect=Exception("Error exporting data to postgres"))
))
async def test_export_data_to_postgres_error(_mock_dataframe, postgres_client):
data = {"schema": "test", "table_name": "test",
"data": {"a": [1, 2, 3], "b": [4, 5, 6]}}
await postgres_client.export_data_to_postgres(data)
# Verify error notification was sent
postgres_client.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES",
message="Error exporting data to postgres: Error exporting data to postgres",
block="export_data_to_postgres",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
# Verify session handling
postgres_client.session_factory.return_value.close.assert_called_once()

82
tests/activities/test_redis.py
View File

@@ -1,5 +1,4 @@
import json
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock, patch, ANY
from datetime import datetime
import pytest
import numpy as np
@@ -9,65 +8,36 @@ from scouter.activities.redis import Redis
@pytest.fixture
@patch('scouter.activities.redis.redis.Redis')
def redis_activity(_mock_redis_client):
@patch('scouter.activities.redis.RedisBase.__init__')
def redis_activity(_mock_redis_init):
logger = MagicMock()
notification_handler = MagicMock(spec=NotificationHandler)
return Redis(host='localhost', port=6379,
logger=logger, notification_handler=notification_handler,
username='test', password='test')
activity = Redis(host='localhost', port=6379,
logger=logger, notification_handler=notification_handler,
username='test', password='test')
activity.redis_client = MagicMock()
activity.logger = logger
activity.notification_handler = notification_handler
return activity
@patch('scouter.activities.redis.redis.Redis')
def test_redis_initialization(mock_redis_client):
@patch('scouter.activities.redis.RedisBase.__init__')
def test_redis_initialization(mock_redis_init):
"""Test Redis activity initialization"""
redis_activity = Redis(host='localhost', port=6379,
logger=MagicMock(), notification_handler=MagicMock(),
username='test', password='test')
assert redis_activity.host == 'localhost'
assert redis_activity.port == 6379
assert redis_activity.username == 'test'
assert redis_activity.password == 'test'
mock_redis_client.assert_called_once_with(
host='localhost',
port=6379,
decode_responses=True,
username='test',
password='test'
)
def test_get_existing_key(redis_activity):
"""Test getting an existing key from Redis"""
test_data = {'key': 'value'}
redis_activity.redis_client.get.return_value = json.dumps(test_data)
result = redis_activity.get('test_key')
assert result == test_data
redis_activity.redis_client.get.assert_called_once_with('test_key')
def test_get_nonexistent_key(redis_activity):
"""Test getting a non-existent key from Redis"""
redis_activity.redis_client.get.return_value = None
result = redis_activity.get('nonexistent_key')
assert result is None
redis_activity.redis_client.get.assert_called_once_with('nonexistent_key')
def test_set_key(redis_activity):
"""Test setting a key in Redis"""
test_data = {'key': 'value'}
redis_activity.set('test_key', test_data, ttl=300)
redis_activity.redis_client.set.assert_called_once_with(
'test_key',
json.dumps(test_data),
ex=300
logger = MagicMock()
notification_handler = MagicMock(spec=NotificationHandler)
Redis(host='localhost', port=6379,
logger=logger, notification_handler=notification_handler,
username='test', password='test')
mock_redis_init.assert_called_once_with(
ANY,
'localhost',
6379,
'test',
'test',
logger,
notification_handler
)

37
tests/utils/test_logger.py
View File

@@ -1,37 +0,0 @@
import os
from unittest.mock import patch
import logging
import pytest
from scouter.utils.logger import get_logger
@pytest.fixture
def mock_env_vars():
with patch.dict(os.environ, {}, clear=True):
yield
@pytest.mark.usefixtures("mock_env_vars")
@patch('scouter.utils.logger.logging.Formatter')
@patch('scouter.utils.logger.logging.StreamHandler')
def test_get_logger_defaults(mock_stream_handler, mock_formatter):
"""Test logger creation with default settings"""
# Mock the StreamHandler and Formatter
logger = get_logger('test_logger')
# Verify logger settings
assert logger.name == 'test_logger'
assert logger.level == logging.INFO
# Verify handler configuration
mock_stream_handler.return_value.setLevel.assert_called_once_with('INFO')
mock_stream_handler.return_value.setFormatter.assert_called_once()
# Verify formatter configuration
mock_formatter.assert_called_once_with(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Verify handler was added to logger
assert len(logger.handlers) == 1

5
values.yaml
View File

@@ -123,7 +123,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1005-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas"
value: "main"
- name: PYTHON_APP
value: "scouter.worker.worker"
@@ -164,7 +164,7 @@ env:
key: redis-password
- name: LOG_LEVEL
value: "INFO"
value: "DEBUG"
- name: PROJECT_NAME
value: "sientia-scouter"
@@ -180,6 +180,7 @@ ssh:
knownHostsPath: /mnt/known_hosts
# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
# helm upgrade --install sientia-scouter-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat
# kubectl create secret generic git-ssh-key-sientia-scouter-worker \