temporal-worker-python

This repository contains TogetherCrew's Temporal Python workflows for data processing and analysis. It leverages the Temporal workflow engine to orchestrate ETL processes and data summarization tasks.

Project Components

Hivemind ETL

• Website Ingestion: Extracts, transforms, and loads data from websites defined in the platform configuration.
• MediaWiki Ingestion: Processes content from MediaWiki instances, including extraction of pages, revisions, and content.

Hivemind Summarizer

• Platform Summaries: Retrieves and processes summaries from Platform data stored in Qdrant, with options to fetch by date or date range.
• Real-Time Summaries: Generates new summaries for recent data across platforms or specific communities.

Architecture

The project uses Temporal for workflow orchestration with the following components:

• Temporal Server: Manages workflow execution and task queues
• MongoDB: Stores platform and community configuration
• Qdrant: Vector database for storing and retrieving summary content
• Redis: Caching and state management
• PostgreSQL: Used by Temporal for workflow history and state

Setup Instructions

1. Configure Environment Variables

   • Copy the example environment file:

   cp .env.example .env

   Update the .env file with your own values, referencing the services defined in docker-compose.dev.yml.

   Required variables:

   • TEMPORAL_TASK_QUEUE: Queue name for the worker
   • Database connection parameters for MongoDB, Qdrant, etc.

2. Start Services

   • Use the following command to set up and run the required services:

   docker compose -f docker-compose.dev.yml up -d

3. Open localhost:8080 to access the Temporal dashboard.

Usage Examples

Running a Platform Summary Workflow

To fetch existing summaries for a specific community and date range from Qdrant:

fromtemporalio.clientimportClientfromhivemind_summarizer.workflowsimportPlatformSummariesWorkflowfromhivemind_summarizer.schemaimportPlatformFetchSummariesWorkflowInputasyncdefrun_platform_summaries_workflow():
client=awaitClient.connect("localhost:7233")
# Create workflow inputinput_data=PlatformFetchSummariesWorkflowInput(
platform_id="your_platform_id", # Required: the platform to fetch summaries fromcommunity_id="your_community_id", # Required: the community to fetch summaries fromstart_date="2023-05-01", # Optional: fetch summaries from this dateend_date="2023-05-07", # Optional: fetch summaries until this dateextract_text_only=True# Optional: whether to extract only text content
)
# Execute workflowresult=awaitclient.execute_workflow(
PlatformSummariesWorkflow.run,
input_data,
id="platform-summaries-workflow",
task_queue="your_task_queue"
)
# Returns a list of existing summaries from Qdrantreturnresult

Note: This workflow only retrieves existing summaries that have already been generated and stored in Qdrant. It does not generate new summaries. Use this when you want to access previously generated summaries for a specific platform and community.

Running a Real-Time Summary Workflow

To generate new summaries for recent data:

fromtemporalio.clientimportClientfromhivemind_summarizer.workflowsimportRealTimeSummaryWorkflowfromhivemind_summarizer.schemaimportRealTimeSummaryWorkflowInputasyncdefrun_realtime_summary_workflow():
client=awaitClient.connect("localhost:7233")
# Create workflow inputinput_data=RealTimeSummaryWorkflowInput(
period="4h", # Optional: time period (e.g., "1h", "4h") or date in %Y-%m-%d formatplatform_id="your_platform_id", # Optional: filter by platformcommunity_id="your_community_id", # Optional: filter by communitycollection_name="your_collection"# Optional: filter by collection
)
# Execute workflowresult=awaitclient.execute_workflow(
RealTimeSummaryWorkflow.run,
input_data,
id="realtime-summary-workflow",
task_queue="your_task_queue"
)
# Returns newly generated summary textreturnresult

Note: This workflow actively generates new summaries for recent data. Use this when you want to create fresh summaries for the specified time period and filters. Note 2: Either one of the filter by collection or filter by platform and community should be given. (to identify the collection to access tha raw data)

Running a MediaWiki ETL Workflow

To process MediaWiki content for all communities or a specific platform:

fromtemporalio.clientimportClientfromhivemind_etl.mediawiki.workflowsimportMediaWikiETLWorkflowasyncdefrun_mediawiki_workflow(platform_id=None):
client=awaitClient.connect("localhost:7233")
# Execute workflow for all platforms or a specific oneawaitclient.execute_workflow(
MediaWikiETLWorkflow.run,
platform_id, # Pass None to process all platformsid="mediawiki-etl-workflow",
task_queue="your_task_queue"
)

Running a Website Ingestion Workflow

To ingest content from websites:

fromtemporalio.clientimportClientfromhivemind_etl.website.workflowsimportWebsiteIngestionSchedulerWorkflowasyncdefrun_website_workflow(platform_id=None):
client=awaitClient.connect("localhost:7233")
# Execute workflow for all communities or a specific oneawaitclient.execute_workflow(
WebsiteIngestionSchedulerWorkflow.run,
platform_id, # Pass None to process all platformsid="website-ingestion-workflow",
task_queue="your_task_queue"
)

Development

To run the worker locally:

python worker.py

This will start a worker that connects to Temporal and listens for tasks on the configured task queue.