Merge pull request #33 from corkrean/temporal-py

adding temporal example

Author: Verolop

data/temporal_example/README.md

@@ -0,0 +1,44 @@
+# Distributed Writes to SpiceDB with Temporal
+
+This is a "quick start" example that demonstrates how you can use Temporal to maintain a consistent state across SpiceDB and an application DB (Postgres in this example) in the event of failures.
+
+Before beginning this example walkthrough, it is recommended to have basic knowledge of Temporal. [Here](https://docs.temporal.io/evaluate/understanding-temporal) is a good place to start with Temporal.
+
+## Prerequisites
+- [Docker Compose](https://docs.docker.com/compose/install/)
+- [Temporal CLI](https://docs.temporal.io/cli#install)
+- [Python 3](https://www.python.org/downloads/)
+- [pip](https://packaging.python.org/en/latest/tutorials/installing-packages/) (included with Python if you installed via Homebrew or python.org) 
+
+## Running this Example
+
+Follow these steps to run this example:
+
+### Setup
+1. Clone the SpiceDB examples repo: ` git clone [email protected]:authzed/examples.git`
+2. Change into the Docker Compose directory `cd examples/data/temporal_example/compose`
+3. Start SpiceDB and Postgres with `docker-compose up -d`
+4. Change to the parent directory: `cd ..`
+5. Start a development Temporal Server: `temporal server start-dev`
+6. Setup a virtual env: `python3 -m venv env`, then `source env/bin/activate`.
+7. Install dependencies: `python3 -m pip install authzed temporalio psycopg`
+8. Setup the Postgres DB and SpiceDB: `python3 run_migrations.py`
+9. Start the Temporal Worker: `python3 run_worker.py`
+
+### Experiencing Temporal
+10. Run the Temporal workflow (worker should still be running while you are doing this): `python3 run_workflow.py --author bob --post some_post`
+11. Simulate a Postgres failure by stopping the Docker container running Postgres: `docker stop dev-postgres`
+12. Run the Temporal workflow (worker should still be running while you are doing this): `python3 run_workflow.py --author bob --post another_post`
+13. Notice that the initial attempt failed.
+14. Restart Postgres: `docker start dev-postgres`
+15. Wait a few seconds and notice that the Postgres write activity succeeds and that the workflow succeeds. 😎
+
+## Important Takeaways from this Example
+
+- This example uses Temporal's default [retry policy](https://docs.temporal.io/encyclopedia/retry-policies) for Activities and Workflows. This means that an Activity will indefinitely try to complete a successful execution.
+
+- Both Postgres and SpiceDB writes in this example are idempotent. In the case of the record already existing, there will be no errors and the workflow will complete successfully. [Temporal recommends that all activities be idempotent.](https://docs.temporal.io/activity-definition#idempotency)
+
+- This example uses a single Temporal Task Queue partition. [Task Queues with a single partition](https://docs.temporal.io/task-queue#task-ordering) are almost always first-in, first-out, with rare edge case exceptions.
+
+- This example writes directly to SpiceDB and Postgres. For many use cases, Temporal will make API requests to distributed microservices that in turn make requests to SpiceDB and an app DB.

data/temporal_example/activities.py

@@ -0,0 +1,22 @@
+from temporalio import activity
+from shared import BlogAuthor
+from data.spicedb import writeRelationship
+from data.postgres import writePostgres
+import psycopg
+
+
+@activity.defn
+async def write_to_spicedb(blog_author: BlogAuthor):
+    try:
+        await writeRelationship(blog_author)
+    except Exception:
+        activity.logger.exception("SpiceDB write failed")
+        raise
+
+@activity.defn
+async def write_to_postgres(blog_author: BlogAuthor):
+    try:
+        await writePostgres(blog_author)
+    except Exception:
+        activity.logger.exception("Postgres write failed")
+        raise

data/temporal_example/compose/docker-compose.yml

@@ -0,0 +1,25 @@
+---
+version: "3.8"
+
+services:
+  postgres:
+    image: "postgres:15"
+    container_name: "dev-postgres"
+    environment:
+      POSTGRES_USER: "dev"
+      POSTGRES_PASSWORD: "abc123"
+      POSTGRES_DB: "blog_authors"
+    ports:
+      - "5432:5432"
+
+  spicedb:
+    image: "authzed/spicedb:v1.34.0"
+    container_name: "dev-spicedb"
+    command: [
+      "serve",
+      "--grpc-preshared-key", "localkey",
+    ]
+    ports:
+      - "50051:50051"   # gRPC
+    depends_on:
+      - "postgres"

data/temporal_example/data/postgres.py

@@ -0,0 +1,31 @@
+import psycopg
+
+from shared import BlogAuthor
+
+DB_CONFIG = {
+    "dbname": "blog_authors",
+    "user": "dev",
+    "password": "abc123",
+    "host": "localhost",
+    "port": 5432
+}
+
+async def writePostgres(blog_author: BlogAuthor):
+    with psycopg.connect(**DB_CONFIG) as conn:
+        with conn.cursor() as cur:
+            cur.execute("""
+                INSERT INTO blog_authors (author_user_id, post_id)
+                VALUES (%s, %s)
+                ON CONFLICT DO NOTHING;
+            """, (blog_author.author_user_id, blog_author.post_id))
+
+async def create_blog_authors_table():
+    with psycopg.connect(**DB_CONFIG) as conn:
+        with conn.cursor() as cur:
+            cur.execute("""
+                CREATE TABLE IF NOT EXISTS blog_authors (
+                    author_user_id TEXT NOT NULL,
+                    post_id TEXT NOT NULL,
+                    PRIMARY KEY (author_user_id, post_id)
+                );
+            """)

data/temporal_example/data/spicedb.py

@@ -0,0 +1,57 @@
+from authzed.api.v1 import (
+        AsyncClient,
+        ObjectReference,
+        Relationship,
+        RelationshipUpdate,
+        SubjectReference,
+        WriteRelationshipsRequest,
+        WriteSchemaRequest,
+    )
+from grpcutil import insecure_bearer_token_credentials
+
+from shared import BlogAuthor
+
+async def writeRelationship(blog_author: BlogAuthor):
+    
+    #Using an insecure bearer token for the example. In a real application, you would use a secure bearer token.
+    client = AsyncClient(
+            "localhost:50051",
+            insecure_bearer_token_credentials("localkey"),
+        )
+    
+    try:
+        await client.WriteRelationships(
+            WriteRelationshipsRequest(
+                updates=[
+                    RelationshipUpdate(
+                        #It's a temporal best practice to keep activities idempotent
+                        operation=RelationshipUpdate.Operation.OPERATION_TOUCH,
+                        relationship=Relationship(
+                            resource=ObjectReference(object_type="post", object_id=blog_author.post_id),
+                            relation="author",
+                            subject=SubjectReference(
+                                object=ObjectReference(
+                                    object_type="user",
+                                    object_id=blog_author.author_user_id,
+                                )
+                            ),
+                        ),
+                    ),
+                ]
+            )
+        )
+    except Exception as e:
+        raise
+
+async def writeSchema():
+    SCHEMA = """definition user {}
+    definition post {
+        relation author: user
+        permission edit = author
+    }"""
+    
+    client = AsyncClient(
+            "localhost:50051",
+            insecure_bearer_token_credentials("localkey"),
+        )
+    await client.WriteSchema(WriteSchemaRequest(schema=SCHEMA))

data/temporal_example/run_migrations.py

@@ -0,0 +1,10 @@
+from data.postgres import create_blog_authors_table
+from data.spicedb import writeSchema
+import asyncio
+
+async def main():
+    await create_blog_authors_table()
+    await writeSchema()
+
+if __name__ == "__main__":
+    asyncio.run(main())

data/temporal_example/run_worker.py

@@ -0,0 +1,24 @@
+import asyncio
+import logging
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from activities import write_to_spicedb, write_to_postgres
+from workflows import writeOwner
+
+async def main():
+    # Configure logging
+    logging.basicConfig(level=logging.INFO)
+
+    client = await Client.connect("localhost:7233", namespace="default")
+    # Run the worker
+    worker = Worker(
+        client, task_queue="write-task-queue", 
+        workflows=[writeOwner], 
+        activities=[write_to_spicedb, write_to_postgres]
+    )
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

data/temporal_example/run_workflow.py

@@ -0,0 +1,26 @@
+import asyncio
+import argparse
+
+from run_worker import writeOwner
+from temporalio.client import Client
+from shared import BlogAuthor
+
+#In a real application, you may invoke this code when someone submits a form, presses a button, or visits a certain URL
+
+async def main(author_user_id: str, post_id: str):
+    client = await Client.connect("localhost:7233")
+
+    await client.execute_workflow(
+        writeOwner.run, BlogAuthor(author_user_id=author_user_id, post_id=post_id), id=f"write-workflow-{author_user_id}-{post_id}", task_queue="write-task-queue"
+    )
+
+    print("Workflow completed successfully")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description='Run the workflow with specified author and post IDs')
+    parser.add_argument('--author', required=True, help='The author user ID')
+    parser.add_argument('--post', required=True, help='The post ID')
+    
+    args = parser.parse_args()
+    asyncio.run(main(args.author, args.post))

data/temporal_example/shared.py

@@ -0,0 +1,6 @@
+from dataclasses import dataclass
+
+@dataclass
+class BlogAuthor:
+    author_user_id: str
+    post_id: str

data/temporal_example/workflows.py

@@ -0,0 +1,40 @@
+from temporalio import workflow
+from datetime import timedelta
+from temporalio.exceptions import ActivityError
+# Import activity, passing it through the sandbox without reloading the module
+with workflow.unsafe.imports_passed_through():
+    from activities import write_to_postgres
+    from activities import write_to_spicedb
+    from shared import BlogAuthor
+    
+
+#this is the workflow definition
+@workflow.defn
+class writeOwner:
+    @workflow.run
+    async def run (self, blog_author: BlogAuthor):
+        try:
+            workflow.logger.info("Starting workflow execution")
+
+            workflow.logger.info("Attempting Postgres write")
+            await workflow.execute_activity(
+                write_to_postgres, 
+                blog_author, 
+                start_to_close_timeout=timedelta(seconds=5),
+            )
+            workflow.logger.info(f"Postgres write completed")
+        except ActivityError as postgresWriteError:
+            workflow.logger.error(f"Postgres write failed (from workflow): {postgresWriteError}")
+            raise postgresWriteError
+
+        try:
+            workflow.logger.info("Attempting SpiceDB write")
+            await workflow.execute_activity(
+                write_to_spicedb, 
+                blog_author, 
+                start_to_close_timeout=timedelta(seconds=5),
+            )
+            workflow.logger.info("SpiceDB write completed")
+        except ActivityError as spicedbWriteError:
+            workflow.logger.error(f"SpiceDB write failed (from workflow): {spicedbWriteError}")
+            raise spicedbWriteError