Service

Bases: ErrorLoom

Service class for handling communication with external APIs.

Source code in blue/service.py

class Service(ErrorLoom):
    """Service class for handling communication with external APIs."""

    def __init__(
        self,
        name="SERVICE",
        id=None,
        sid=None,
        cid=None,
        prefix=None,
        suffix=None,
        handler=None,
        properties={},
    ):
        """Initialize the Service.

        Parameters:
            name: Name of the service. Defaults to "SERVICE".
            id: Unique identifier for the service. Defaults to None.
            sid: Short identifier for the service. Defaults to None.
            cid: Canonical identifier for the service. Defaults to None.
            prefix: Optional prefix for the cid. Defaults to None.
            suffix: Optional suffix for the cid. Defaults to None.
            handler: Callback function to handle service requests. Defaults to None.
            properties: Additional properties for the service. Defaults to {}.
        """
        self.name = name
        if id:
            self.id = id
        else:
            self.id = uuid_utils.create_uuid()

        if sid:
            self.sid = sid
        else:
            self.sid = self.name + ":" + self.id

        self.prefix = prefix
        self.suffix = suffix
        self.cid = cid

        if self.cid == None:
            self.cid = self.sid

            if self.prefix:
                self.cid = self.prefix + ":" + self.cid
            if self.suffix:
                self.cid = self.cid + ":" + self.suffix

        self._initialize(properties=properties)

        # override, if necessary
        _handler = handler if handler is not None else self.default_handler

        # Preserve async nature when injecting properties
        if asyncio.iscoroutinefunction(_handler):
            async def async_wrapper(*args, **kwargs):
                return await _handler(*args, **kwargs, properties=self.properties)
            self.handler = async_wrapper
        else:
            self.handler = partial(_handler, properties=self.properties)

        self._start()

    ###### initialization
    def _initialize(self, properties=None):
        """Initialize the service with properties.

        Parameters:
            properties: Additional properties for the service. Defaults to None.
        """
        self._initialize_properties()
        self._update_properties(properties=properties)

        self._initialize_logger()

    def _initialize_properties(self):
        """Initialize default properties for the service."""
        self.properties = {}

        # db connectivity
        self.properties["db.host"] = "localhost"
        self.properties["db.port"] = 6379

        # stats tracker
        self.properties["tracker.perf.service.autostart"] = True
        self.properties["tracker.perf.service.outputs"] = ["pubsub"]

    def _update_properties(self, properties=None):
        """Update service properties with provided properties.

        Parameters:
            properties: Additional properties for the service. Defaults to None.
        """
        if properties is None:
            return

        # override
        for p in properties:
            self.properties[p] = properties[p]

    def _initialize_logger(self):
        """Initialize the logger for the service."""
        self.logger = log_utils.CustomLogger()
        # customize log
        self.logger.set_config_data(
            "stack",
            "%(call_stack)s",
        )
        self.logger.set_config_data("service", self.sid, -1)

    ###### database, data
    def _start_connection(self):
        """Start the database connection using a pooled connection factory."""
        self.connection_factory = PooledConnectionFactory(properties=self.properties)
        self.connection = self.connection_factory.get_connection()

    ##### tracker
    def stat_tracker_callback(self, data, tracker=None, properties=None):
        """Callback function for service performance tracking."""
        pass

    def _init_tracker(self):
        """Initialize the service performance tracker."""
        # service stat tracker
        self._tracker = ServicePerformanceTracker(self, properties=self.properties, callback=lambda *args, **kwargs: self.stat_tracker_callback(*args, **kwargs))

    def _start_tracker(self):
        """Start the service performance tracker."""
        # start tracker
        self._tracker.start()

    def _stop_tracker(self):
        """Stop the service performance tracker."""
        self._tracker.stop()

    def _terminate_tracker(self):
        """Terminate the service performance tracker."""
        self._tracker.terminate()

    ## service metadata
    def __get_json_value(self, value):
        if value is None:
            return None
        if type(value) is list:
            if len(value) == 0:
                return None
            else:
                return value[0]
        else:
            return value

    def _init_metadata_namespace(self):
        """Initialize the metadata namespace for the service, sets created_date and initializes key stats for websockets and total call count."""
        # create namespaces for metadata
        self.connection.json().set(
            self._get_metadata_namespace(),
            "$",
            {"stats": {}},
            nx=True,
        )

        # add created_date
        self.set_metadata("created_date", int(time.time()), nx=True)

        # websockers
        self.set_metadata("stats.websockets", {}, nx=True)

        # total call count
        self.set_metadata("stats.total_call_count", int(0), nx=True)

    def _get_metadata_namespace(self):
        """Get the metadata namespace for the service."""
        return self.cid + ":METADATA"

    def set_metadata(self, key, value, nx=False):
        """Set metadata for the service.

        Parameters:
            key: Metadata key to set.
            value: Value to set for the metadata key.
            nx: If True, set the value only if the key does not already exist. Defaults to False.
        """
        self.connection.json().set(self._get_metadata_namespace(), "$." + key, value, nx=nx)

    def delete_metadata(self, key):
        """Delete metadata for the service.

        Parameters:
            key: Metadata key to delete.
        """
        self.connection.json().delete(self._get_metadata_namespace(), "$." + key)

    def get_metadata(self, key=""):
        """Get metadata for the service.

        Parameters:
            key: Metadata key to retrieve. Defaults to "".
        Returns:
            Value of the metadata key, or None if the key does not exist.
        """
        value = self.connection.json().get(
            self._get_metadata_namespace(),
            Path("$" + ("" if pydash.is_empty(key) else ".") + key),
        )
        return self.__get_json_value(value)

    def _init_socket_stats(self, websocket):
        """Initialize socket statistics for a given websocket connection.

        Parameters:
            websocket: WebSocket connection object.
        """
        # stats by websocket.id
        wsid = websocket.id
        self.set_metadata("stats.websockets." + str(wsid), {}, nx=True)

        self.set_socket_stat(websocket, "created_date", int(time.time()), nx=True)

    def set_socket_stat(self, websocket, key, value, nx=False):
        """Set a specific statistic for a given websocket connection.

        Parameters:
            websocket: WebSocket connection object.
            key: Statistic key to set.
            value: Value to set for the statistic key.
            nx: If True, set the value only if the key does not already exist. Defaults to False.
        """
        wsid = websocket.id
        self.set_metadata("stats.websockets." + str(wsid) + "." + key, value, nx=True)

    def error_handler(self, error: BlueError, exception: Exception):
        context = error.context
        websocket_to_respond = pydash.objects.get(context, 'websocket', None)
        if websocket_to_respond:

            async def send_error_message():
                if hasattr(websocket_to_respond, 'open') and not websocket_to_respond.open:
                    return
                await websocket_to_respond.send(json.dumps({"status": "server_error", "error": error.get_dict()}))

            def run_in_thread(loop):
                asyncio.set_event_loop(loop)
                try:
                    loop.run_until_complete(send_error_message())
                except Exception as e:
                    threading.current_thread().exception = e

            new_loop = asyncio.new_event_loop()
            thread = threading.Thread(target=run_in_thread, args=(new_loop,))
            thread.daemon = True
            thread.start()
            thread.join(timeout=10)

    ###### handlers
    async def _handler(self, websocket):
        """Handle incoming WebSocket messages and process them using the service's handler function.
        Sets up socket statistics and processes messages in a loop until the connection is closed.

        Parameters:
            websocket: WebSocket connection object.
        """
        try:
            self._init_socket_stats(websocket)

            while True:
                try:
                    ### read message
                    s = await websocket.recv()

                    # message length
                    self.set_socket_stat(websocket, "length", len(s))

                    message = json.loads(s)

                    ### process message
                    start = time.time()
                    # Check if handler is async and await if necessary
                    if asyncio.iscoroutinefunction(self.handler):
                        response = await self.handler(message, websocket=websocket)
                    else:
                        response = self.handler(message, websocket=websocket)
                    end = time.time()
                    self.set_socket_stat(websocket, "response_time", end - start)

                    ### write response
                    await websocket.send(json.dumps(response))

                except websockets.ConnectionClosedOK:
                    break
        except Exception as ex:
            error = BlueError(ex, context={'websocket': websocket})
            raise error

    async def start_listening_socket(self):
        """Start listening for incoming WebSocket connections on port 8001."""
        async with websockets.serve(self._handler, "", 8001):
            await asyncio.Future()  # run forever

    ## default handler, override
    @skip_error_loom
    def default_handler(self, message, properties=None, websocket=None):
        """Default handler for processing incoming messages. This method should be overridden by subclasses to implement custom behavior.

        Parameters:
            message: Incoming message to process.
            properties: Additional properties for the handler. Defaults to None.
            websocket: WebSocket connection object. Defaults to None.
        Returns:
            Response message. Should be overridden to provide meaningful responses.
        """
        self.logger.info("default_handler: override")

    def _start(self):
        """Start the service by establishing a database connection, initializing metadata, and starting the performance tracker if configured."""
        self._start_connection()

        # initialize session metadata
        self._init_metadata_namespace()

        # init tracker
        self._init_tracker()

        self.logger.info("Started service {name}".format(name=self.name))

    def stop(self):
        """Stop the service by stopping and terminating the performance tracker."""
        self.logger.info("Stopped servie {name}".format(name=self.name))

__init__(name='SERVICE', id=None, sid=None, cid=None, prefix=None, suffix=None, handler=None, properties={})

Initialize the Service.

Parameters:

Name	Type	Description	Default
name		Name of the service. Defaults to "SERVICE".	'SERVICE'
id		Unique identifier for the service. Defaults to None.	None
sid		Short identifier for the service. Defaults to None.	None
cid		Canonical identifier for the service. Defaults to None.	None
prefix		Optional prefix for the cid. Defaults to None.	None
suffix		Optional suffix for the cid. Defaults to None.	None
handler		Callback function to handle service requests. Defaults to None.	None
properties		Additional properties for the service. Defaults to {}.	{}

Source code in blue/service.py

def __init__(
    self,
    name="SERVICE",
    id=None,
    sid=None,
    cid=None,
    prefix=None,
    suffix=None,
    handler=None,
    properties={},
):
    """Initialize the Service.

    Parameters:
        name: Name of the service. Defaults to "SERVICE".
        id: Unique identifier for the service. Defaults to None.
        sid: Short identifier for the service. Defaults to None.
        cid: Canonical identifier for the service. Defaults to None.
        prefix: Optional prefix for the cid. Defaults to None.
        suffix: Optional suffix for the cid. Defaults to None.
        handler: Callback function to handle service requests. Defaults to None.
        properties: Additional properties for the service. Defaults to {}.
    """
    self.name = name
    if id:
        self.id = id
    else:
        self.id = uuid_utils.create_uuid()

    if sid:
        self.sid = sid
    else:
        self.sid = self.name + ":" + self.id

    self.prefix = prefix
    self.suffix = suffix
    self.cid = cid

    if self.cid == None:
        self.cid = self.sid

        if self.prefix:
            self.cid = self.prefix + ":" + self.cid
        if self.suffix:
            self.cid = self.cid + ":" + self.suffix

    self._initialize(properties=properties)

    # override, if necessary
    _handler = handler if handler is not None else self.default_handler

    # Preserve async nature when injecting properties
    if asyncio.iscoroutinefunction(_handler):
        async def async_wrapper(*args, **kwargs):
            return await _handler(*args, **kwargs, properties=self.properties)
        self.handler = async_wrapper
    else:
        self.handler = partial(_handler, properties=self.properties)

    self._start()

default_handler(message, properties=None, websocket=None)

Default handler for processing incoming messages. This method should be overridden by subclasses to implement custom behavior.

Parameters:

Name	Type	Description	Default
message		Incoming message to process.	required
properties		Additional properties for the handler. Defaults to None.	None
websocket		WebSocket connection object. Defaults to None.	None

Returns: Response message. Should be overridden to provide meaningful responses.

Source code in blue/service.py

@skip_error_loom
def default_handler(self, message, properties=None, websocket=None):
    """Default handler for processing incoming messages. This method should be overridden by subclasses to implement custom behavior.

    Parameters:
        message: Incoming message to process.
        properties: Additional properties for the handler. Defaults to None.
        websocket: WebSocket connection object. Defaults to None.
    Returns:
        Response message. Should be overridden to provide meaningful responses.
    """
    self.logger.info("default_handler: override")

delete_metadata(key)

Delete metadata for the service.

Parameters:

Name	Type	Description	Default
key		Metadata key to delete.	required

Source code in blue/service.py

def delete_metadata(self, key):
    """Delete metadata for the service.

    Parameters:
        key: Metadata key to delete.
    """
    self.connection.json().delete(self._get_metadata_namespace(), "$." + key)

get_metadata(key='')

Get metadata for the service.

Parameters:

Name	Type	Description	Default
key		Metadata key to retrieve. Defaults to "".	''

Returns: Value of the metadata key, or None if the key does not exist.

Source code in blue/service.py

def get_metadata(self, key=""):
    """Get metadata for the service.

    Parameters:
        key: Metadata key to retrieve. Defaults to "".
    Returns:
        Value of the metadata key, or None if the key does not exist.
    """
    value = self.connection.json().get(
        self._get_metadata_namespace(),
        Path("$" + ("" if pydash.is_empty(key) else ".") + key),
    )
    return self.__get_json_value(value)

set_metadata(key, value, nx=False)

Set metadata for the service.

Parameters:

Name	Type	Description	Default
key		Metadata key to set.	required
value		Value to set for the metadata key.	required
nx		If True, set the value only if the key does not already exist. Defaults to False.	False

Source code in blue/service.py

def set_metadata(self, key, value, nx=False):
    """Set metadata for the service.

    Parameters:
        key: Metadata key to set.
        value: Value to set for the metadata key.
        nx: If True, set the value only if the key does not already exist. Defaults to False.
    """
    self.connection.json().set(self._get_metadata_namespace(), "$." + key, value, nx=nx)

set_socket_stat(websocket, key, value, nx=False)

Set a specific statistic for a given websocket connection.

Parameters:

Name	Type	Description	Default
websocket		WebSocket connection object.	required
key		Statistic key to set.	required
value		Value to set for the statistic key.	required
nx		If True, set the value only if the key does not already exist. Defaults to False.	False

Source code in blue/service.py

def set_socket_stat(self, websocket, key, value, nx=False):
    """Set a specific statistic for a given websocket connection.

    Parameters:
        websocket: WebSocket connection object.
        key: Statistic key to set.
        value: Value to set for the statistic key.
        nx: If True, set the value only if the key does not already exist. Defaults to False.
    """
    wsid = websocket.id
    self.set_metadata("stats.websockets." + str(wsid) + "." + key, value, nx=True)

start_listening_socket() async

Start listening for incoming WebSocket connections on port 8001.

Source code in blue/service.py

async def start_listening_socket(self):
    """Start listening for incoming WebSocket connections on port 8001."""
    async with websockets.serve(self._handler, "", 8001):
        await asyncio.Future()  # run forever

stat_tracker_callback(data, tracker=None, properties=None)

Callback function for service performance tracking.

Source code in blue/service.py

def stat_tracker_callback(self, data, tracker=None, properties=None):
    """Callback function for service performance tracking."""
    pass

stop()

Stop the service by stopping and terminating the performance tracker.

Source code in blue/service.py

def stop(self):
    """Stop the service by stopping and terminating the performance tracker."""
    self.logger.info("Stopped servie {name}".format(name=self.name))

Last update: 2025-10-08