-
Notifications
You must be signed in to change notification settings - Fork 18
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add python event processing examples
- Loading branch information
Showing
16 changed files
with
196 additions
and
173 deletions.
There are no files selected for viewing
This file was deleted.
Oops, something went wrong.
This file was deleted.
Oops, something went wrong.
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| version: '3' | ||
| services: | ||
| broker: | ||
| image: confluentinc/cp-kafka:7.5.0 | ||
| container_name: broker | ||
| ports: | ||
| - "9092:9092" | ||
| - "9101:9101" | ||
| environment: | ||
| KAFKA_BROKER_ID: 1 | ||
| KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT | ||
| KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092 | ||
| KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 | ||
| KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 | ||
| KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 | ||
| KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 | ||
| KAFKA_PROCESS_ROLES: broker,controller | ||
| KAFKA_NODE_ID: 1 | ||
| KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093 | ||
| KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092 | ||
| KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT | ||
| KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER | ||
| KAFKA_LOG_DIRS: /tmp/kraft-combined-logs | ||
| CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk | ||
|
|
||
| init-kafka: | ||
| image: confluentinc/cp-kafka:7.5.0 | ||
| depends_on: | ||
| - broker | ||
| entrypoint: [ '/bin/sh', '-c' ] | ||
| command: | | ||
| "# blocks until kafka is reachable | ||
| kafka-topics --bootstrap-server broker:29092 --list | ||
| echo -e 'Creating kafka topics' | ||
| kafka-topics --bootstrap-server broker:29092 --create --if-not-exists --topic package-location-updates --replication-factor 1 --partitions 1 | ||
| echo -e 'Successfully created the following topics:' | ||
| kafka-topics --bootstrap-server broker:29092 --list" |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| [[ingress.kafka-clusters]] | ||
| name = "my-cluster" | ||
| brokers = ["PLAINTEXT://localhost:9092"] |
Empty file.
This file contains 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
Empty file.
55 changes: 55 additions & 0 deletions
55
python/patterns-use-cases/src/eventenrichment/package_tracker.py
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| from typing import List | ||
|
|
||
| import restate | ||
| from pydantic import BaseModel | ||
| from restate import VirtualObject, ObjectContext | ||
| from restate.exceptions import TerminalError | ||
|
|
||
|
|
||
| class LocationUpdate(BaseModel): | ||
| location: str | ||
| timestamp: str | ||
|
|
||
|
|
||
| class PackageInfo(BaseModel): | ||
| final_destination: str | ||
| locations: List[LocationUpdate] = [] | ||
|
|
||
|
|
||
| # Package tracking system: | ||
| # Digital twin representing a package in delivery with real-time location updates. | ||
| # Handlers get called over HTTP or Kafka. | ||
| package_tracker = VirtualObject("package-tracker") | ||
|
|
||
|
|
||
| # Called first by the seller over HTTP | ||
| @package_tracker.handler("registerPackage") | ||
| async def register_package(ctx: ObjectContext, package_info: PackageInfo): | ||
| # store in state the user's information as coming from the registration event | ||
| ctx.set("package-info", package_info.model_dump()) | ||
|
|
||
|
|
||
| # Connected to a Kafka topic for real-time location updates | ||
| @package_tracker.handler("updateLocation") | ||
| async def update_location(ctx: ObjectContext, location_update: LocationUpdate): | ||
| # get the package info from the state | ||
| package_info = PackageInfo(**await ctx.get("package-info")) | ||
| if package_info is None: | ||
| raise TerminalError(f"Package {ctx.key()} not found") | ||
|
|
||
| # Update the package details in the state | ||
| locations = package_info.locations or [] | ||
| locations.append(location_update) | ||
| package_info.locations = locations | ||
|
|
||
| # store the updated package info in state | ||
| ctx.set("package-info", package_info.model_dump()) | ||
|
|
||
|
|
||
| # Called by the delivery dashboard to get the package details | ||
| @package_tracker.handler("getPackageInfo") | ||
| async def get_package_info(ctx: ObjectContext) -> PackageInfo: | ||
| return PackageInfo(**await ctx.get("package-info")) | ||
|
|
||
|
|
||
| app = restate.app(services=[package_tracker]) |
Empty file.
37 changes: 37 additions & 0 deletions
37
python/patterns-use-cases/src/eventtransactions/user_feed.py
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| import restate | ||
| from restate import VirtualObject, ObjectContext | ||
| from datetime import timedelta | ||
|
|
||
| from src.eventtransactions.utils import create_post, get_post_status, update_user_feed, SocialMediaPost | ||
|
|
||
| # Processing events (from Kafka) to update various downstream systems | ||
| # - Journaling actions in Restate and driving retries from Restate, recovering | ||
| # partial progress | ||
| # - Preserving the order-per-key, but otherwise allowing high-fanout, because | ||
| # processing of events does not block other events. | ||
| # - Ability to delay events when the downstream systems are busy, without blocking | ||
| # entire partitions. | ||
| user_feed = VirtualObject("UserFeed") | ||
|
|
||
|
|
||
| # The Kafka key routes events to the correct Virtual Object. | ||
| # Events with the same key are processed one after the other. | ||
| @user_feed.handler("processPost") | ||
| async def process_post(ctx: ObjectContext, post: SocialMediaPost): | ||
| user_id = ctx.key() | ||
|
|
||
| # event handler is a durably executed function that can use all the features of Restate | ||
| post_id = await ctx.run("profile update", lambda: create_post(user_id, post)) | ||
|
|
||
| # Delay processing until content moderation is complete (handler suspends when on FaaS). | ||
| # This only blocks other posts for this user (Virtual Object), not for other users. | ||
| while await ctx.run("post status", lambda: get_post_status(post_id)) == "PENDING": | ||
| await ctx.sleep(timedelta(seconds=5)) | ||
|
|
||
| await ctx.run("update feed", lambda: update_user_feed(user_id, post_id)) | ||
|
|
||
|
|
||
| app = restate.app(services=[user_feed]) | ||
|
|
||
| # Process new posts for users via Kafka or by calling the endpoint over HTTP: curl | ||
| # localhost:8080/userFeed/userid1/processPost --json '{"content": "Hi! This is my first post!", "metadata": "public"}' |
Oops, something went wrong.