totoms 1.0.1

Toto Microservice SDK - Python framework for building microservices

Toto Microservice SDK - Python

Python framework for building cloud-agnostic microservices with FastAPI.

Table of Contents

1. Installation
2. Overview
3. Usage
   • 3.1. Create and Register APIs
   • 3.2. Use a Message Bus
   • 3.3. Load Secrets
   • 3.4. Custom Configurations
4. Reference: Toto Microservice Configuration
5. License

1. Installation

pip install totoms

2. Overview

Everything starts with TotoMicroservice.
TotoMicroservice is the main orchestrator that coordinates your entire microservice. It initializes and manages:

• API Controller & API Endpoints: FastAPI-based REST API setup with automatic endpoint registration
• Message Bus & Message Handlers: Event-driven communication via Pub/Sub and Queues. Registration and routing of event handlers to appropriate topics.
• Secrets Management: Automatic loading of secrets from your cloud provider
• Service Lifecycle: Initialization, startup, and shutdown management

The configuration is declarative - you create a TotoMicroserviceConfiguration object that specifies:

• Service Metadata: Service name and base path for API endpoints
• Environment: Cloud provider (AWS, GCP, Azure) information
• API Configuration: REST endpoints with their handlers
• Message Bus Configuration: Topics to subscribe to and message handlers
• Custom Configuration: Your application-specific settings

The microservice initialization is async and returns a fully configured instance ready to start:

microservice = await TotoMicroservice.init(get_microservice_config())
await microservice.start(port=8080)

3. Usage

3.1. Create and Register APIs

Your microservice exposes REST API endpoints using FastAPI.
Endpoints are defined when creating the microservice configuration and are automatically set up with the API controller.

Create a Toto Delegate

Every endpoint needs to be managed by a Toto Delegate.
Toto Delegates are identified through the annotation @toto_delegate.

This is how you define a Toto Delegate.
The following example shows a delegate that extracts the content of a blog.

@toto_delegate
async def extract_blog_content(request: Request, user_context: UserContext, exec_context: ExecutionContext): 

    # Extract and parse the body from FastAPI Request
    body = await request.json()

    blog_url = data.get("url")
    
    # Process the blog
    result = await process_blog(blog_url)
    
    # Return anything you'd like to
    return {"status": "success", "data": result}

Register your delegate

You can now register your endpoints in the microservice configuration:

def get_microservice_config() -> TotoMicroserviceConfiguration:
    
    return TotoMicroserviceConfiguration(
        service_name="my-service",
        base_path="/myservice",
        api_configuration=APIConfiguration(
            api_endpoints=[
                APIEndpoint(method="POST", path="/blogs", delegate=extract_blog_content),
            ]
        ),
    )

The microservice will start a FastAPI application with all registered endpoints available at the specified base path.

3.2. Use a Message Bus

The Message Bus enables event-driven communication between microservices.
It supports both PUSH (webhook-based from cloud Pub/Sub) and PULL (polling) delivery models, depending on your cloud provider and configuration.

3.2.1. React to Messages

Message handlers are the primary way to react to events.

Create a Message Handler

Create a handler by subclassing TotoMessageHandler and implementing the required methods:

from totoms import TotoMessageHandler, ProcessingResponse, ProcessingStatus, TotoMessage

class TopicRefreshedEventHandler(TotoMessageHandler):
    
    def get_handled_message_type(self) -> str:
        """Return the message type this handler processes."""
        return "topicRefreshed"
    
    async def process_message(self, message: TotoMessage) -> ProcessingResponse:
        """Process incoming messages of type 'topicRefreshed'."""
        # Access message metadata
        correlation_id = message.cid
        message_id = message.id
        
        # Extract event data
        topic_name = message.data.get("name")
        blog_url = message.data.get("blogURL")
        user = message.data.get("user")
        
        # Your handler has access to context
        self.logger.log(correlation_id, f"Processing topic refresh for: {topic_name}")
        
        # Perform your business logic
        await refresh_topic(topic_name, blog_url, user)
        
        # Return success or failure
        return ProcessingResponse(status=ProcessingStatus.SUCCESS)

Register a Message Handler

Register your message handlers in the microservice configuration.

IMPORTANT NOTE:

• When using PubSub infrastructure, you need to register topics.
  Topics are registered by giving them:
  • A logical_name which is the name that will be used in the application to reference the topic.
  • A secret which is the name of the secret that contains the implementation-specific resource identifier of the topic (e.g. ARN on AWS or fully-qualified Topic Name on GCP)

from totoms import MessageBusHandlerConfig

def get_microservice_config() -> TotoMicroserviceConfiguration:
    """Create configuration with message handlers."""
    return TotoMicroserviceConfiguration(
        service_name="my-service",
        message_bus_configuration=MessageBusConfig(
            topics=[
                MessageBusTopicConfig(
                    logical_name="topic-events", 
                    secret="topic_events_topic_name"  # Secret name in your secrets manager
                )
            ],
            message_handlers=[
                MessageBusHandlerConfig(handler_class=TopicRefreshedEventHandler),
                MessageBusHandlerConfig(handler_class=AnotherEventHandler),
            ]
        ),
    )

When the microservice starts, it automatically subscribes to the configured topics and routes incoming messages to the appropriate handlers based on their message type.

3.2.2. Publish Messages

You can always publish messages to topics.

NOTE:

• In the Message Destination, the topic is the logical name of the topic (see above).

from totoms import TotoMessage, MessageDestination

async def publish_topic_update(microservice, topic_id: str, topic_name: str):
    """Publish a topic update event."""
    message = TotoMessage(
        type="topicUpdated",
        cid="correlation-id-123",
        id=topic_id,
        data={"name": topic_name, "timestamp": datetime.now().isoformat()}
    )
    
    destination = MessageDestination(topic="topic-events")

    await microservice.message_bus.publish_message(destination, message)

Getting access to the Message Bus.

There are different ways to get access to the Message Bus instance:

• Through the TotoMicroservice singleton:
  TotoMicroservice.get_instance().message_bus

• Through an existing instance of TotoMicroservice (in the example above)

• In a TotoMessageHandler you will have message_bus as an instance variable:
  self.message_bus

• In a toto_delegate, you will have it part of ExecutionContext and can use like this:
  exec_context.message_bus

3.3. Load Secrets

The SDK handles secret loading from your cloud provider automatically. Access secrets through the configuration or use the SecretsManager directly:

from totoms import SecretsManager

secrets = SecretsManager()

# Load a secret by name
api_key = secrets.get_secret("api-key")
database_url = secrets.get_secret("database-url")

Secrets are typically stored as environment variable names or secret manager references, depending on your deployment environment.

3.4. Custom Configurations

You can define your own custom configurations by extending the TotoControllerConfig base class.

An example:

from totoms.model.TotoConfig import TotoControllerConfig
from typing import Optional, Dict

class TomeScraperConfig(TotoControllerConfig):
    """Custom configuration for the Toto Tome Scraper service."""
    
    def get_mongo_secret_names(self) -> Optional[Dict[str, str]]:
        """Return MongoDB secret names if service uses MongoDB."""
        return None

What you can do with a Custom Configuration:

1. Load Secrets
   You can do that by overriding the load() async method and using self.secrets_manager.get_secret, "your-secret-name") to load secrets.

4. Reference: Toto Microservice Configuration

This is a full example of an app.py root that configures a TotoMicroservice.

import asyncio
import os
from config.config import TomeScraperConfig
from evt.handlers.TopicRefreshedMH import TopicRefreshedEventHandler
from totoms import (
    MessageBusHandlerConfig,
    TotoMicroservice,
    TotoMicroserviceConfiguration,
    TotoEnvironment,
    APIConfiguration,
)
from totoms.TotoMicroservice import APIEndpoint, determine_environment, MessageBusTopicConfig, MessageBusConfig

from dlg.scrape import extract_blog_content
...

def get_microservice_config() -> TotoMicroserviceConfiguration:
    
    return TotoMicroserviceConfiguration(
        service_name="toto-ms-tome-scraper",
        base_path="/tomescraper",
        environment=TotoEnvironment(
            hyperscaler=os.getenv("HYPERSCALER", "aws").lower(),
            hyperscaler_configuration=determine_envrionment()
        ),
        custom_config=TomeScraperConfig,
        api_configuration=APIConfiguration(
            api_endpoints=[
                APIEndpoint(method="POST", path="/blogs", delegate=extract_blog_content),
                APIEndpoint(method="POST", path="/test/refresher", delegate=test_refresher),
                APIEndpoint(method="POST", path="/test/pubsub", delegate=test_pubsub),
            ]
        ),
        message_bus_configuration=MessageBusConfig(
            topics=[
                MessageBusTopicConfig(logical_name="tometopics", secret="tome_topics_topic_name")
            ], 
            message_handlers=[
                MessageBusHandlerConfig(handler_class=TopicRefreshedEventHandler)
            ]
        ),
    )


async def main():
    
    microservice = await TotoMicroservice.init(get_microservice_config())

    port = int(os.getenv("PORT", "8080"))
    
    await microservice.start(port=port)


if __name__ == "__main__":
    asyncio.run(main())

The following fields are defined in TotoMicroserviceConfiguration:

• service_name - logical name of your microservice. Mostly used for logging.
• base_path - Very Important - add this to have your microservice exposed on a base path. It's very useful for routing.
  Example: /tomescraper as a basepath will mean that your service will be exposed on <whatever-domain-you-have-configured>/tomescraper/<....your routes>
• environment - defines in which environment your service will run:
  • hyperscaler - defines whether it runs on AWS, GCP or Azure
  • hyperscaler_configuration - defines hyperscaler-specific configurations (e.g. region for AWS, project id for GCP, etc...).
    The function determine_envrionment() is a utility function that automatically determines the environment configuration based on Environment Variables
• custom_config - defines any custom configuration you have. This is a mandatory field.

License

MIT License - see LICENSE file for details