Merge pull request #594 from avison9/feature/parsingVesu

[DataHandler] Add background task for parsing Vesu

Author: djeck1432

apps/shared/background_tasks/data_handler/event_tasks.py

@@ -3,14 +3,44 @@
 """
 
 import logging
+import asyncio
 from datetime import datetime
+import threading
 
 from apps.shared.celery_conf import app
 from data_handler.handlers.events.nostra.transform_events import NostraTransformer
 from data_handler.handlers.events.zklend.transform_events import ZklendTransformer
+from apps.data_handler.handlers.loan_states.vesu.events import VesuLoanEntity
 
 logger = logging.getLogger(__name__)
 
+CHUNK_SIZE = 5
+
+"""in the event of mulitple loop run async method in a new thread to avoid event loop conflicts"""
+def run_async_in_thread(coro):
+    """Run an async coroutine in a new thread with its own event loop."""
+    result = None
+    exception = None
+
+    def run_coro():
+        nonlocal result, exception
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            result = loop.run_until_complete(coro)
+        except Exception as e:
+            exception = e
+        finally:
+            loop.close()
+
+    thread = threading.Thread(target=run_coro)
+    thread.start()
+    thread.join()
+
+    if exception:
+        raise exception
+    return result
+
 
 @app.task(name="process_zklend_events")
 def process_zklend_events():
@@ -96,3 +126,44 @@ def process_nostra_events():
             exc,
             exc_info=True,
         )
+
+
+@app.task(name="process_vesu_events", bind=True, max_retries=3)
+def process_vesu_events(self):
+    """
+    Process and store Vesu protocol events.
+    Fetches ModifyPosition events from the blockchain, updates user positions,
+    and stores them in a mock database.
+    """
+    start_time = datetime.utcnow()
+    logger.info("Starting Vesu event processing")
+    try:
+        vesu_entity = VesuLoanEntity()
+        run_async_in_thread(vesu_entity.update_positions_data())
+        execution_time = (datetime.utcnow() - start_time).total_seconds()
+        logger.info(
+            "Successfully processed Vesu events in %.2fs (UTC). Blocks: %d to %d",
+            execution_time, 
+            vesu_entity.last_processed_block - CHUNK_SIZE, # default as update_positions_data chunk_size
+            vesu_entity.last_processed_block,
+        )
+    except (ValueError, TypeError, RuntimeError) as exc:
+        execution_time = (datetime.utcnow() - start_time).total_seconds()
+        logger.error(
+            "Error processing Vesu events after %.2fs: %s",
+            execution_time,
+            exc,
+            exc_info=True,
+        )
+        self.retry(countdown=60)
+    except Exception as exc:
+        execution_time = (datetime.utcnow() - start_time).total_seconds()
+        logger.error(
+            "Unexpected error processing Vesu events after %.2fs: %s",
+            execution_time,
+            exc,
+            exc_info=True,
+        )
+        self.retry(countdown=60)
+
+

apps/shared/celery_conf.py

@@ -67,6 +67,10 @@
         "task": "process_nostra_events",
         "schedule": crontab(minute=f"*/{CRONTAB_TIME}"),
     },
+    "process_vesu_events_every_5_mins": {
+        "task": "process_vesu_events",
+        "schedule": crontab(minute="*/5"),
+    },
 }
 
 from apps.shared.background_tasks.data_handler.order_books_tasks import ekubo_order_book

apps/shared/pyproject.toml

@@ -23,6 +23,7 @@ psycopg2 = "^2.9.10"
 sqlalchemy-utils = "^0.41.2"
 pydantic-settings = "^2.8.1"
 asyncpg = "^0.30.0"
+celery = "^5.3"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.3"
@@ -31,6 +32,7 @@ black = "^23.3.0"
 isort = "^5.12.0"
 mypy = "^1.3.0"
 flake8 = "^6.0.0"
+celery = "^5.3"
 
 
 

apps/shared/tests/test_process_vesu_event.py

@@ -0,0 +1,113 @@
+# import pytest
+# import time
+# import asyncio
+# from unittest.mock import Mock, AsyncMock, patch
+# from datetime import datetime
+# import logging
+# from apps.data_handler.handlers.loan_states.vesu.events import VesuLoanEntity
+# from apps.shared.background_tasks.data_handler.event_tasks import process_vesu_events 
+
+
+# logging.basicConfig(level=logging.INFO)
+# logger = logging.getLogger(__name__)
+
+# @pytest.fixture
+# def mock_vesu_entity():
+#     """Fixture to mock VesuLoanEntity."""
+#     mock = Mock(spec=VesuLoanEntity)
+#     mock.last_processed_block = 654244  # Initial block
+#     return mock
+
+# @pytest.fixture
+# def caplog(caplog):
+#     """Fixture to capture log output."""
+#     caplog.set_level(logging.INFO, logger="apps.shared.background_tasks.data_handler.event_tasks")
+#     return caplog
+
+# def test_process_vesu_events_success(mock_vesu_entity, caplog, monkeypatch):
+#     """Test successful execution of process_vesu_events."""
+#     # Mock VesuLoanEntity Class
+#     monkeypatch.setattr(
+#         "apps.shared.background_tasks.data_handler.event_tasks.VesuLoanEntity",
+#         Mock(return_value=mock_vesu_entity)
+#     )
+#     # Mock update_positions_data
+#     async def mock_update():
+#         mock_vesu_entity.last_processed_block = 654249
+#     mock_vesu_entity.update_positions_data = AsyncMock(side_effect=mock_update)
+
+#     process_vesu_events()
+
+#     assert "Starting Vesu event processing" in caplog.text
+#     assert "Successfully processed Vesu events" in caplog.text
+#     assert "Blocks: 654244 to 654249" in caplog.text
+#     assert "(UTC)" in caplog.text
+#     assert mock_vesu_entity.last_processed_block == 654249
+
+# def test_process_vesu_events_value_error(mock_vesu_entity, caplog, monkeypatch):
+#     """Test handling of ValueError."""
+#     monkeypatch.setattr(
+#         "apps.shared.background_tasks.data_handler.event_tasks.VesuLoanEntity",
+#         Mock(return_value=mock_vesu_entity)
+#     )
+#     mock_vesu_entity.update_positions_data = AsyncMock(side_effect=ValueError("Failed to fetch events from Starknet"))
+#     # Mock retry to prevent Retry exception
+#     with patch.object(process_vesu_events, "retry", side_effect=Exception("Retry mocked")):
+#         with pytest.raises(Exception, match="Retry mocked"):
+#             process_vesu_events()
+
+#     assert "Starting Vesu event processing" in caplog.text
+#     assert "Error processing Vesu events" in caplog.text
+#     assert "Failed to fetch events from Starknet" in caplog.text
+
+# def test_process_vesu_events_unexpected_error(mock_vesu_entity, caplog, monkeypatch):
+#     """Test handling of unexpected errors."""
+#     monkeypatch.setattr(
+#         "apps.shared.background_tasks.data_handler.event_tasks.VesuLoanEntity",
+#         Mock(return_value=mock_vesu_entity)
+#     )
+#     mock_vesu_entity.update_positions_data = AsyncMock(side_effect=Exception("Unexpected Starknet error"))
+#     # Mock retry to prevent Retry exception
+#     with patch.object(process_vesu_events, "retry", side_effect=Exception("Retry mocked")):
+#         with pytest.raises(Exception, match="Retry mocked"):
+#             process_vesu_events()
+
+#     assert "Starting Vesu event processing" in caplog.text
+#     assert "Unexpected error processing Vesu events" in caplog.text
+#     assert "Unexpected Starknet error" in caplog.text
+
+# def test_process_vesu_events_scheduled(mock_vesu_entity, caplog, monkeypatch):
+#     """Test periodic execution of process_vesu_events with a 3-second interval."""
+#     chunk_size = 5
+#     # Mock VesuLoanEntity
+#     monkeypatch.setattr(
+#         "apps.shared.background_tasks.data_handler.event_tasks.VesuLoanEntity",
+#         Mock(return_value=mock_vesu_entity)
+#     )
+
+#     async def mock_update():
+#         mock_vesu_entity.last_processed_block += chunk_size
+#     mock_vesu_entity.update_positions_data = AsyncMock(side_effect=mock_update)
+
+#     # Simulate 3 runs with 10-second intervals
+#     initial_block = mock_vesu_entity.last_processed_block
+#     num_runs = 3
+#     for i in range(num_runs):
+#         process_vesu_events()
+#         if i < num_runs - 1:  
+#             time.sleep(10)
+
+
+#     expected_blocks = [
+#         (initial_block, initial_block + chunk_size),
+#         (initial_block + chunk_size, initial_block + 2 * chunk_size),
+#         (initial_block + 2 * chunk_size, initial_block + 3 * chunk_size)
+#     ]
+#     for start_block, end_block in expected_blocks:
+#         assert f"Blocks: {start_block} to {end_block}" in caplog.text
+
+
+#     assert mock_vesu_entity.last_processed_block == initial_block + num_runs * chunk_size
+#     assert "Starting Vesu event processing" in caplog.text
+#     assert "Successfully processed Vesu events" in caplog.text
+#     assert "(UTC)" in caplog.text