From 82a0c7dc783344c80c4f5d6703c1b97109bf9d81 Mon Sep 17 00:00:00 2001 From: Matan <51418643+matan84@users.noreply.github.com> Date: Tue, 31 Dec 2024 15:01:12 +0200 Subject: [PATCH] [Core] Create webhooks only resync (#1225) # Description What - Added new event listener type- Webhooks only Why - In order to enable 2 different integration types How - Added new event listener without resync logic ## Type of change Please leave one option from the following and delete the rest: - [ ] Bug fix (non-breaking change which fixes an issue) - [x] New feature (non-breaking change which adds functionality) - [ ] New Integration (non-breaking change which adds a new integration) - [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected) - [ ] Non-breaking change (fix of existing functionality that will not change current behavior) - [ ] Documentation (added/updated documentation)

All tests should be run against the port production environment(using a testing org).

### Core testing checklist - [ ] Integration able to create all default resources from scratch - [ ] Resync finishes successfully - [ ] Resync able to create entities - [ ] Resync able to update entities - [ ] Resync able to detect and delete entities - [ ] Scheduled resync able to abort existing resync and start a new one - [ ] Tested with at least 2 integrations from scratch - [ ] Tested with Kafka and Polling event listeners - [ ] Tested deletion of entities that don't pass the selector ### Integration testing checklist - [ ] Integration able to create all default resources from scratch - [ ] Resync able to create entities - [ ] Resync able to update entities - [ ] Resync able to detect and delete entities - [ ] Resync finishes successfully - [ ] If new resource kind is added or updated in the integration, add example raw data, mapping and expected result to the `examples` folder in the integration directory. - [ ] If resource kind is updated, run the integration with the example data and check if the expected result is achieved - [ ] If new resource kind is added or updated, validate that live-events for that resource are working as expected - [ ] Docs PR link [here](#) ### Preflight checklist - [ ] Handled rate limiting - [ ] Handled pagination - [ ] Implemented the code in async - [ ] Support Multi account ## Screenshots Include screenshots from your environment showing how the resources of the integration will look. ## API Documentation Provide links to the API documentation used for this integration. --------- Co-authored-by: Tom Tankilevitch <59158507+Tankilevitch@users.noreply.github.com> --- CHANGELOG.md | 8 ++++ port_ocean/context/ocean.py | 15 +++++-- port_ocean/core/event_listener/__init__.py | 8 ++++ port_ocean/core/event_listener/base.py | 1 + port_ocean/core/event_listener/factory.py | 10 ++++- .../core/event_listener/webhooks_only.py | 41 +++++++++++++++++++ port_ocean/core/integrations/base.py | 6 ++- port_ocean/run.py | 1 - pyproject.toml | 2 +- 9 files changed, 85 insertions(+), 7 deletions(-) create mode 100644 port_ocean/core/event_listener/webhooks_only.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f3e579291..1b9cc13fc6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## 0.17.0 (2024-12-31) + + +### Features + +- Added new webhooks only event listener mode. This event listener handles only webhook invocations and raises error once used for resync. + + ## 0.16.1 (2024-12-25) ### Bug Fixes diff --git a/port_ocean/context/ocean.py b/port_ocean/context/ocean.py index 0830729788..c9d37f37db 100644 --- a/port_ocean/context/ocean.py +++ b/port_ocean/context/ocean.py @@ -23,6 +23,8 @@ from port_ocean.ocean import Ocean from port_ocean.clients.port.client import PortClient +from loguru import logger + class PortOceanContext: def __init__(self, app: Union["Ocean", None]) -> None: @@ -63,14 +65,21 @@ def port_client(self) -> "PortClient": return self.app.port_client @property - def event_listener_type(self) -> Literal["WEBHOOK", "KAFKA", "POLLING", "ONCE"]: + def event_listener_type( + self, + ) -> Literal["WEBHOOK", "KAFKA", "POLLING", "ONCE", "WEBHOOKS_ONLY"]: return self.app.config.event_listener.type def on_resync( self, kind: str | None = None, - ) -> Callable[[RESYNC_EVENT_LISTENER], RESYNC_EVENT_LISTENER]: - def wrapper(function: RESYNC_EVENT_LISTENER) -> RESYNC_EVENT_LISTENER: + ) -> Callable[[RESYNC_EVENT_LISTENER], RESYNC_EVENT_LISTENER | None]: + def wrapper(function: RESYNC_EVENT_LISTENER) -> RESYNC_EVENT_LISTENER | None: + if not self.app.config.event_listener.should_resync: + logger.debug( + "Webhook only event listener is used, resync events are ignored" + ) + return None return self.integration.on_resync(function, kind) return wrapper diff --git a/port_ocean/core/event_listener/__init__.py b/port_ocean/core/event_listener/__init__.py index 4e26dab725..d9939b5322 100644 --- a/port_ocean/core/event_listener/__init__.py +++ b/port_ocean/core/event_listener/__init__.py @@ -16,12 +16,18 @@ OnceEventListener, ) +from port_ocean.core.event_listener.webhooks_only import ( + WebhooksOnlyEventListener, + WebhooksOnlyEventListenerSettings, +) + EventListenerSettingsType = ( HttpEventListenerSettings | KafkaEventListenerSettings | PollingEventListenerSettings | OnceEventListenerSettings + | WebhooksOnlyEventListenerSettings ) __all__ = [ @@ -34,4 +40,6 @@ "PollingEventListenerSettings", "OnceEventListener", "OnceEventListenerSettings", + "WebhooksOnlyEventListener", + "WebhooksOnlyEventListenerSettings", ] diff --git a/port_ocean/core/event_listener/base.py b/port_ocean/core/event_listener/base.py index 04c43d87f1..9c9821ccf8 100644 --- a/port_ocean/core/event_listener/base.py +++ b/port_ocean/core/event_listener/base.py @@ -76,6 +76,7 @@ async def _resync( class EventListenerSettings(BaseOceanModel, extra=Extra.allow): type: str + should_resync: bool = True def to_request(self) -> dict[str, Any]: """ diff --git a/port_ocean/core/event_listener/factory.py b/port_ocean/core/event_listener/factory.py index 6127dfc05a..c3c8d68110 100644 --- a/port_ocean/core/event_listener/factory.py +++ b/port_ocean/core/event_listener/factory.py @@ -17,6 +17,10 @@ BaseEventListener, EventListenerEvents, ) +from port_ocean.core.event_listener.webhooks_only import ( + WebhooksOnlyEventListener, + WebhooksOnlyEventListenerSettings, +) from port_ocean.exceptions.core import UnsupportedEventListenerTypeException @@ -88,7 +92,11 @@ async def create_event_listener(self) -> BaseEventListener: config, OnceEventListenerSettings ), assert_message.format(type(config)) event_listener = OnceEventListener(wrapped_events, config) - + case "webhooks_only": + assert isinstance( + config, WebhooksOnlyEventListenerSettings + ), assert_message.format(type(config)) + event_listener = WebhooksOnlyEventListener(wrapped_events, config) case _: raise UnsupportedEventListenerTypeException( f"Event listener {_type} not supported" diff --git a/port_ocean/core/event_listener/webhooks_only.py b/port_ocean/core/event_listener/webhooks_only.py new file mode 100644 index 0000000000..c55afc5b39 --- /dev/null +++ b/port_ocean/core/event_listener/webhooks_only.py @@ -0,0 +1,41 @@ +from typing import Literal + +from loguru import logger + +from port_ocean.core.event_listener.base import ( + BaseEventListener, + EventListenerEvents, + EventListenerSettings, +) + + +class WebhooksOnlyEventListenerSettings(EventListenerSettings): + """ + This class inherits from `EventListenerSettings`, which provides a foundation for creating event listener settings. + """ + + type: Literal["WEBHOOKS_ONLY"] + should_resync: bool = False + + +class WebhooksOnlyEventListener(BaseEventListener): + """ + No resync event listener. + + It is used to handle events exclusively through webhooks without supporting resync events. + + Parameters: + events (EventListenerEvents): A dictionary containing event types and their corresponding event handlers. + event_listener_config (OnceEventListenerSettings): The event listener configuration settings. + """ + + def __init__( + self, + events: EventListenerEvents, + event_listener_config: WebhooksOnlyEventListenerSettings, + ): + super().__init__(events) + self.event_listener_config = event_listener_config + + async def _start(self) -> None: + logger.info("Starting Webhooks-only event listener") diff --git a/port_ocean/core/integrations/base.py b/port_ocean/core/integrations/base.py index fa14bf91f4..df3ce08bca 100644 --- a/port_ocean/core/integrations/base.py +++ b/port_ocean/core/integrations/base.py @@ -54,13 +54,17 @@ async def start(self) -> None: """ Initializes handlers, establishes integration at the specified port, and starts the event listener. """ - logger.info("Starting integration", integration_type=self.context.config.integration.type) + logger.info( + "Starting integration", + integration_type=self.context.config.integration.type, + ) if self.started: raise IntegrationAlreadyStartedException("Integration already started") if ( not self.event_strategy["resync"] and self.__class__._on_resync == BaseIntegration._on_resync + and self.context.config.event_listener.should_resync ): raise NotImplementedError("on_resync is not implemented") diff --git a/port_ocean/run.py b/port_ocean/run.py index b90f27f816..c6d349ca1f 100644 --- a/port_ocean/run.py +++ b/port_ocean/run.py @@ -57,7 +57,6 @@ def run( # Override config with arguments if initialize_port_resources is not None: app.config.initialize_port_resources = initialize_port_resources - initialize_defaults(app.integration.AppConfigHandlerClass.CONFIG_CLASS, app.config) uvicorn.run(app, host="0.0.0.0", port=application_settings.port) diff --git a/pyproject.toml b/pyproject.toml index f6c02bd788..d991bc34d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "port-ocean" -version = "0.16.1" +version = "0.17.0" description = "Port Ocean is a CLI tool for managing your Port projects." readme = "README.md" homepage = "https://app.getport.io"