drop pipeline events, functionality moved to nacsos-data

.gitignore

@@ -7,4 +7,4 @@ __pycache__
 /hypercorn.error
 .pyc
 server.md
-dump.sql
\ No newline at end of file
+dumps/
\ No newline at end of file

requirements.txt

@@ -2,11 +2,11 @@ fastapi==0.92.0
 hypercorn==0.14.3
 toml==0.10.2
 email-validator==1.3.1
-python-dotenv==0.21.1
+python-dotenv==1.0.0
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
 httpx[http2]==0.23.3
 pymitter==0.4.0
 uvicorn==0.20.0
 python-multipart==0.0.5
-nacsos_data @ git+ssh://git@gitlab.pik-potsdam.de/mcc-apsis/nacsos/nacsos-data.git@v0.2.3#egg=nacsos_data
+nacsos_data @ git+ssh://git@gitlab.pik-potsdam.de/mcc-apsis/nacsos/nacsos-data.git@v0.3.0#egg=nacsos_data

requirements_dev.txt

 flake8==6.0.0
-tox==4.4.5
+tox==4.4.6
 pytest==7.2.1
 pytest-cov==4.0.0
 mypy==1.0.1
-types-toml==0.10.8.4
\ No newline at end of file
+types-toml==0.10.8.5
\ No newline at end of file

server/util/events/__init__.py

@@ -2,7 +2,7 @@ from typing import Union, Literal, TYPE_CHECKING
 
 from pymitter import EventEmitter
 
-from .hooks import imports
+from .hooks import example
 from . import events
 
 eventbus = EventEmitter(delimiter='_', wildcard=True)
@@ -16,6 +16,6 @@ else:
     AnyEventLiteral = Literal[tuple(sc._name for sc in events.BaseEvent.get_subclasses())]  # noqa PyProtectedMember
 
 # Permanent/global listeners
-eventbus.on(events.PipelineTaskStatusChangedEvent._name, imports.update_import_status)  # noqa PyProtectedMember
+eventbus.on(events.ExampleEvent._name, example.test_listener)  # noqa PyProtectedMember
 
 __all__ = ['eventbus', 'events', 'AnyEvent', 'AnyEventType']

server/util/events/events.py

-from typing import Literal, ClassVar
+from typing import ClassVar
 
 from pydantic import BaseModel
 
@@ -17,38 +17,10 @@ class BaseEvent(BaseModel):
         return tuple(set(recurse(cls)))
 
 
-TaskStatus = Literal['PENDING', 'RUNNING', 'COMPLETED', 'FAILED', 'CANCELLED']  # FIXME move somewhere else
+class ExampleEvent(BaseEvent):
+    _name = 'Example_*'
+    payload_a: str
 
 
-class PipelineTaskStatusChangedEvent(BaseEvent):
-    """
-    Emitted when the pipeline service calls the nacsos-core API and tells it about a status change of a task
-    """
-    _name = 'PipelineTaskStatus_*'
-    task_id: str
-    status: TaskStatus
-    project_id: str
-    user_id: str
-    import_id: str | None = None
-    function_name: str  # incl module path
-
-
-class PipelineTaskStatusCompletedEvent(PipelineTaskStatusChangedEvent):
-    """
-    More specific version of `PipelineTaskStatusChangedEvent` emitted when a task finished (successfully/completed)
-    """
-    _name = 'PipelineTask_completed'
-
-
-class PipelineTaskStatusStartedEvent(PipelineTaskStatusChangedEvent):
-    """
-    More specific version of `PipelineTaskStatusChangedEvent` emitted when a task started
-    """
-    _name = 'PipelineTask_started'
-
-
-class PipelineTaskStatusFailedEvent(PipelineTaskStatusChangedEvent):
-    """
-    More specific version of `PipelineTaskStatusChangedEvent` emitted when a task finished (failed)
-    """
-    _name = 'PipelineTask_failed'
+class ExampleSubEvent(ExampleEvent):
+    _name = 'Example_sub'

server/util/events/hooks/example.py

+import logging
+
+from ..events import ExampleEvent
+
+logger = logging.getLogger('nacsos.event-hooks.test')
+
+
+def test_listener(event: ExampleEvent):
+    logger.debug(f'Received event {event}')

server/util/events/hooks/imports.py

-import logging
-from datetime import datetime
-
-from sqlalchemy import select
-from nacsos_data.db.schemas import Import
-
-from ..events import PipelineTaskStatusChangedEvent
-from ....data import db_engine
-
-logger = logging.getLogger('nacsos.event-hooks.imports')
-
-
-async def update_import_status(event: PipelineTaskStatusChangedEvent):
-    logger.debug(f'Maybe going to update import status {event}')
-    if event.function_name in [
-        'nacsos_lib.twitter.import.import_twitter_api',
-        'nacsos_lib.twitter.import.import_twitter_db',
-        'nacsos_lib.academic.import.import_wos_file'
-    ]:
-        async with db_engine.session() as session:
-
-            stmt = select(Import).filter_by(pipeline_task_id=event.task_id)
-            import_details: Import | None = (await session.execute(stmt)).scalars().one_or_none()
-
-            if import_details is None and event.import_id is not None:
-                logger.debug(f'second try with {event.import_id}')
-                stmt = select(Import).filter_by(import_id=event.import_id)
-                import_details = (await session.execute(stmt)).scalars().one_or_none()
-
-            if import_details is not None:
-
-                # Seems like task was started, remember the time
-                if event.status == 'RUNNING' and import_details.time_started is None:
-                    logger.debug(  # type: ignore[unreachable]
-                        f'Updating import start time for {import_details.import_id}'
-                    )
-                    import_details.time_started = datetime.now()
-                    await session.commit()
-
-                elif (event.status == 'COMPLETED' or event.status == 'FAILED' or event.status == 'CANCELLED') \
-                        and import_details.time_finished is None:
-                    logger.debug(  # type: ignore[unreachable]
-                        f'Updating import finish time for {import_details.import_id}'
-                    )
-                    import_details.time_finished = datetime.now()
-                    await session.commit()