APP-9587 : Using Pubsub for message processors #898
+829
−3
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
📜 Docstring Coverage ReportRESULT: PASSED (minimum: 30.0%, actual: 76.6%) Detailed Coverage Report |
📦 Trivy Vulnerability Scan Results
Report Summary
Scan Result Details✅ No vulnerabilities found during the scan for |
📦 Trivy Secret Scan Results
Report Summary
Scan Result Details✅ No secrets found during the scan for |
|
🛠 Docs available at: https://k.atlan.dev/application-sdk/add-message-processor |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Collaborator
☂️ Python Coverage
Overall Coverage
New FilesNo new covered files... Modified FilesNo covered modified files...
|
|
🛠 Full Test Coverage Report: https://k.atlan.dev/coverage/application-sdk/pr/898 |
niteesh-atlan
changed the title
niteesh-atlan
requested review from
AtMrun,
firecast and
inishchith
as code owners
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces Pub/Sub messaging support for message processors using Dapr, enabling applications to subscribe to message topics and process incoming messages through configurable handlers.
Key changes:
- Added Dapr pubsub component configuration with support for both in-memory and Kafka backends
- Introduced
PubSubSubscriptionandBulkSubscribemodels for configuring message subscriptions - Implemented automatic registration of message handler endpoints and Dapr subscription generation
Reviewed changes
Copilot reviewed 3 out of 4 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
| components/messaging.yaml | New Dapr pubsub component configuration file with in-memory and commented Kafka configurations |
| components/eventstore.yaml | Minor formatting fix to ensure both webhook URLs remain commented out |
| application_sdk/server/fastapi/models.py | Added BulkSubscribe and PubSubSubscription models to support message subscription configuration |
| application_sdk/server/fastapi/init.py | Integrated messaging subscriptions into server initialization, router registration, and Dapr subscription endpoint |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
niteesh-atlan
changed the title
3 tasks
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Signed-off-by: niteesh-atlan <niteesh.hegde@atlan.com>
firecast
reviewed
Jan 6, 2026
Member
firecast
left a comment
firecast
left a comment
There was a problem hiding this comment.
We also need docs on how can the handlers be written and a typedefinition defined for it as arguments so that handlers are specific
| subscription.message_handler, | ||
| methods=["POST"], | ||
| ) | ||
| self.app.include_router(messaging_router, prefix="/message-processor") |
Member
There was a problem hiding this comment.
Lets have the prefix as subscriptions
Member
There was a problem hiding this comment.
Do we also need to version this path? Check how events are being handled
Author
There was a problem hiding this comment.
updated it. makes sense to add v1 if we standardize event-structure / cloud events format.
Author
There was a problem hiding this comment.
@amit-atlan - you are talking about these docs only right? I could not find these details readme in sdk.
I will add the developer docs and link sample apps here.
Author
There was a problem hiding this comment.
Added developer docs - https://github.com/atlanhq/atlan-docs/pull/512.
Details about dapr component - bulk and single message processors, DLQ and retries
niteesh-atlan
requested review from
abhishekagrawal-atlan,
Copilot and
firecast
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 5 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (1)
application_sdk/server/fastapi/init.py:127
- The docstring for the init method is incomplete. It only documents three parameters (lifespan, handler, workflow_client) but the method signature includes additional parameters: frontend_templates_path, ui_enabled, has_configmap, and the newly added messaging_subscriptions. All parameters should be documented in the Args section of the docstring for clarity and maintainability.
"""Initialize the FastAPI application.
Args:
lifespan: Optional lifespan manager for the FastAPI application.
handler: Handler for processing application operations.
workflow_client: Client for Temporal workflow operations.
"""
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| workflow_router: APIRouter | ||
| dapr_router: APIRouter | ||
| events_router: APIRouter | ||
| messaging_router: APIRouter |
There was a problem hiding this comment.
The newly added messaging_router attribute is declared but never initialized or used. Messaging routes are registered using a local variable instead (see line 226). Consider removing this unused attribute or updating the implementation to use it consistently.
Suggested change
| messaging_router: APIRouter |
firecast
requested changes
Jan 8, 2026
| self.dapr_router = APIRouter() | ||
| self.events_router = APIRouter() | ||
|
|
||
| if len(messaging_subscriptions) > 0: |
| """ | ||
|
|
||
| subscriptions: List[dict[str, Any]] = [] | ||
| if self.messaging_subscriptions: |
| "route": f"/subscriptions/v1/{subscription.route}", | ||
| } | ||
| if subscription.bulk_subscribe and subscription.bulk_subscribe.enabled: | ||
| subscription_dict["bulkSubscribe"] = { |
Member
There was a problem hiding this comment.
just do ** since the keys are the same
| if subscription.bulk_subscribe and subscription.bulk_subscribe.enabled: | ||
| subscription_dict["bulkSubscribe"] = { | ||
| "enabled": subscription.bulk_subscribe.enabled, | ||
| "maxMessagesCount": subscription.bulk_subscribe.maxMessagesCount, |
| maxAwaitDurationMs: int = 40 | ||
|
|
||
|
|
||
| class PubSubSubscription(BaseModel): |
|
|
||
| model_config = {"arbitrary_types_allowed": True} | ||
|
|
||
| pubsub_component_name: str |
| model_config = {"arbitrary_types_allowed": True} | ||
|
|
||
| pubsub_component_name: str | ||
| topic: str |
Member
There was a problem hiding this comment.
might not be required if component is always tied to a topic
| Callable[[Any], Any], Callable[[Any], Coroutine[Any, Any, Any]] | ||
| ] # Required callback function (sync or async) | ||
| bulk_subscribe: Optional[BulkSubscribe] = None | ||
| dead_letter_topic: Optional[str] = None |
| message_handler: Union[ | ||
| Callable[[Any], Any], Callable[[Any], Coroutine[Any, Any, Any]] | ||
| ] # Required callback function (sync or async) | ||
| bulk_subscribe: Optional[BulkSubscribe] = None |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Changelog
APP-9587 : Using Pubsub for message processors
Scope
Sample app Pr - atlanhq/atlan-sample-apps#119
Additional context (e.g. screenshots, logs, links)
Checklist
Copyleft License Compliance