BOSA Core

BOSA Core is a flexible plugin-based architecture that allows you to build modular and extensible applications. It provides a robust framework for managing plugins, handling dependencies, and injecting services.

Key Features

• 🔌 Plugin-based Architecture
• 💉 Dependency Injection
• 🔄 Service Registry
• 🛠️ Flexible Plugin Handlers
• 🌐 HTTP Interface Support
• 📊 OpenTelemetry Integration
• 🛡️ Sentry Error Tracking
• 📜 Logging

Installation

Prerequisites

• Python 3.11+ - Install here
• Pip (if using Pip) - Install here
• Poetry 2.1.3+ (if using Poetry) - Install here
• Git (if using Git) - Install here
• For git installation:
  • Access to the GDP Labs SDK github repository

1. Installation from Pypi

Choose one of the following methods to install the package:

Using pip

pip install bosa-core-binary

Using Poetry

poetry add bosa-core-binary

2. Development Installation (Git)

For development purposes, you can install directly from the Git repository:

poetry add "git+ssh://git@github.com/GDP-ADMIN/bosa-sdk.git#subdirectory=python/bosa-core"

Plugin Architecture Overview

How It Works

1. Plugin Manager: The central piece that orchestrates everything

   • Manages plugin lifecycle
   • Handles service registration
   • Initializes plugins with required dependencies

2. Service Registry: A container for all services

   • Stores service instances
   • Handles dependency injection
   • Maps service types to their implementations
   • Note: You don't need to create your own Service Registry - it's integrated into the Manager!

3. Plugin Handler: Interface for providing services to plugins

   • Creates service injections through create_injections
   • Initializes plugin-specific resources
   • Can be extended for different types of plugins (e.g., HTTP handlers)

4. Plugin: Base class for all plugins

   • Has basic metadata (name, description, version)
   • Receives injected services automatically
   • Can be extended with specific functionality

5. Open Telemetry: Initializer class to use open telemetry

   • Integrates with OpenTelemetry for tracing and metrics
   • Provides observability for plugins and services
   • Can be configured to export data to various backends (e.g., Jaeger, Prometheus)

6. Sentry: Initializer class to use sentry

   • Integrates with Sentry for error tracking and performance monitoring
   • Captures exceptions and sends them to Sentry
   • Provides insights into application performance and errors

Authentication

BOSA Core provides a robust, multi-tenancy authentication system, allowing secure client and user management.

1. Client Management:

   • Clients are created using a Master Key (BOSA_WHITELISTED_KEYS).
   • Each client has a unique API key used for authentication.
   • API Key Format: sk-client-<base64-client-id>.<base64-client-name>.<client-secret>.
   • Clients can manage their users and authentication settings.

2. User Authentication:

   • Clients create users by providing an identifier.
   • Each user receives a secret, which is displayed only once.
   • Users authenticate using their identifier and secret.
   • User API Key Format: sk-user-<base64-user-id>.<base64-user-identifier>.<user-secret>.

3. Token-Based Access:

   • Upon authentication, users receive JWT tokens for secure access.
   • Tokens are verified for subsequent requests.
   • Tokens expire after 30 days (default: 43,200 minutes).
   • Tokens can be revoked when necessary.

4. Third-Party Integrations:

   • Users can create integrations with third-party services.
   • Authentication credentials are securely stored.
   • Each integration can be assigned scoped access.

For detailed implementation, refer to the Authentication README.

Implementation Details

The plugin system works through these key components:

1. Plugin Manager

from fastapi_interface import FastApiHttpInterface

# Create FastAPI handler
fastapi_handler = FastApiHttpInterface()

# Initialize manager with the handler
manager = PluginManager(handlers=[fastapi_handler])
manager.register_plugin(MyPlugin)

• Initializes the plugin system
• Manages plugin lifecycle
• Handles service registration

2. Plugin Handler

from bosa_core.plugin import PluginHandler

class MyHandler(PluginHandler):
    @classmethod
    def create_injections(cls, instance):
        # Tell the manager what services this handler provides
        return {MyService: instance}

    @classmethod
    def initialize_plugin(cls, instance, plugin):
        # Initialize plugin resources
        pass

• Creates service mappings via create_injections
• Initializes plugin resources
• Can be customized for different plugin types

3. Plugin

from bosa_core.plugin import Plugin
from fastapi_interface import FastApiService

class MyPlugin(Plugin):
    # Services will be injected automatically
    fastapi_service: FastApiService

    def __init__(self, name, description, version):
        super().__init__(name, description, version)

• Base class for all plugins
• Receives injected services
• Can be extended with custom functionality

4. Open telemetry external exporter (enable fastapi & langchain instrumentation)

from bosa_core.telemetry import OpenTelemetryConfig, FastAPIConfig, init_telemetry
from fastapi import FastAPI

endpoint = "your-exporter-endpoint"
port = "your-exporter-port"
attributes = {
   "key1":"val1"
}
app = FastAPI()
use_langchain_instrumentation = True
fastapi_config = FastAPIConfig(app=app)
otel_config = OpenTelemetryConfig(endpoint=endpoint, port=port, attributes=attributes, fastapi_config=fastapi_config, use_langchain=use_langchain_instrumentation)
init_telemetry(TelemetryConfig(otel_config=otel_config))

5. Sentry

from bosa_core.telemetry import SentryConfig, init_telemetry

def traces_sampler(sampling_context: dict) -> float:
    """Determine appropriate sampling rate for Sentry transactions.
    Args:
        sampling_context: Context dictionary containing transaction information
    Returns:
        float: Sampling rate between 0 and 1
    """

    ### you can use customization (ex: for certain name sampler will be 0.1) for trace_sampler in this class
    return 1.0


dsn = "your-dsn"
environment = "development/staging/production"
release = "your release tag"
sentry_config = SentryConfig(
        dsn=dsn,
        environment=environment,
        release=release,
        traces_sampler=traces_sampler,
        send_default_pii=True,
    )
init_telemetry(TelemetryConfig(sentry_config=sentry_config))

6. Sentry with Open Telemetry (enable fastapi & langchain instrumentation)

from bosa_core.telemetry import SentryConfig, init_telemetry, FastAPIConfig, OpenTelemetryConfig
from fastapi import FastAPI


def traces_sampler(sampling_context: dict) -> float:
    """Determine appropriate sampling rate for Sentry transactions.
    Args:
        sampling_context: Context dictionary containing transaction information
    Returns:
        float: Sampling rate between 0 and 1
    """

    ### you can use customization (ex: for certain name sampler will be 0.1) for trace_sampler in this class
    return 1.0


dsn = "your-dsn"
environment = "development/staging/production"
release = "your release tag"

attributes = {
   "key1":"val1"
}
app = FastAPI()
use_langchain_instrumentation = True
fastapi_config = FastAPIConfig(app=app)
opentelemetry_init = OpenTelemetryConfig(attributes=attributes, fastapi_config=fastapi_config, use_langchain=use_langchain_instrumentation)

sentry_config = SentryConfig(
        dsn=dsn,
        environment=environment,
        release=release,
        traces_sampler=traces_sampler,
        send_default_pii=True,
        open_telemetry_initiliazer=opentelemetry_init
    )
init_telemetry(sentry_config=sentry_config)

7. Logging Ner API Handler

from bosa_core.logger import init_ner_pii_logging_handler

init_ner_pii_logging_handler(logger_name="logging-name", api_url="https://dev-api-gdplabs-ner-api.obrol.id/anonymize", api_field="text", pii_ner_process_enabled=True)


logger.info("contoh nomor ktp 3525011212941001\ncontoh email john.doe@example.com\ncontoh nomor telepon +628121729819 dan 0812898029384.\ncontoh npwp 01.123.456.7-891.234" ) # if you using gdplabs ner API output will be contoh nomor ktp <ID_KTP_1>\ncontoh email <EMAIL_ADDRESS_1>\ncontoh nomor telepon <PHONE_NUMBER_1> dan <PHONE_NUMBER_2>.\ncontoh npwp <ID_NPWP_1>

8. Logging Regex Handler

from bosa_core.logger import init_regex_pii_logging_handler

init_regex_pii_logging_handler(logger_name="logging-name", pii_regex_process_enabled=True)


logger.info("contoh nomor ktp 3525011212941001\ncontoh email john.doe@example.com\ncontoh nomor telepon +628121729819 dan 0812898029384.\ncontoh npwp 01.123.456.7-891.234" ) # if you using gdplabs ner API output will be contoh nomor ktp <ID_KTP_1>\ncontoh email <EMAIL_ADDRESS_1>\ncontoh nomor telepon <PHONE_NUMBER_1> dan <PHONE_NUMBER_2>.\ncontoh npwp <ID_NPWP_1>

Getting Started

1. Install dependencies:

   poetry install
   

2. Create a plugin:

   class MyPlugin(Plugin):
       name = "my-plugin"
       description = "My awesome plugin"
       version = "1.0.0"
   

3. Register and use:

   # Using FastAPI as an example
   from fastapi import FastAPI
   from fastapi_interface import FastApiHttpInterface
   
   app = FastAPI()
   fastapi_handler = FastApiHttpInterface(app)
   manager = PluginManager(handlers=[fastapi_handler])
   manager.register_plugin(MyPlugin)
   

Development

• Use Poetry for dependency management
• Install pre-commit hooks poetry run pre-commit install. This will run checks before each commit.
• Run ./build.local.sh to perform all checks:
  • Code formatting and validation (ruff)
  • Documentation style (pydocstyle)
  • Tests (pytest)
  • And more!

References

Product Requirements Documents(PRD):

• N/A

Architecture Documents:

• BOSA Core - Architecture Document

Design Documents:

• BOSA Core - Design Document

Implementation Documents:

• BOSA Core - Implementation Document