bosa-core-binary 0.5.3

BOSA Core Library

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://[e-mail]/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 [e-mail]\ncontoh nomor telepon [phone] 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>

[!WARNING] If you are using Logger Ner API handler, the logging process may take some time because it needs to call an API for every log entry synchronously.

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 [e-mail]\ncontoh nomor telepon [phone] 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.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