Getting started

To start using core-service you need to install it using pip:

pip install -U core-service

Then you need to import core_service.Service and subclass it first:

from core_service import Service

class MyService(Service):
    pass

Such service can be started and stopped already:

import asyncio

async def main():
    service = MyService()
    await service.start()
    await service.stop()

asyncio.run(main())

You should always stop started service.

Service task

There is a service task concept. Service task is a asyncio task that is managed by the service. Example of such task may be a polling of data etc.

Failed task will fail the whole service.

Service task can be defined as an async method of Service class marked with core_service.task() decorator.

from core_service import Service, task


class MyService(Service):
    @task(sleep_interval=1)
    async def my_task(self):
        self.log.info("my_task executed)

Task will be executed in infinite loop and will print log message every 1 sec. Also, you can run multiple tasks in parallel by providing workers parameter greater than 1. Sleeping interval will be applied to each instance individually. See core_service.task() reference for details.

Nested services

Service can contain multiple nested services inside. Such services will be started and stopped with their parent. Failed service will fail his parent and all siblings too. Nested services will be started after service tasks were running.

Nested services can be defined as async methods returning list of nested service instances and wrapped with core_service.requirements() decorator.

from core_service import Service, requirements

class NestedService(Service):
    pass

class MyService(Service):
    @requirements()
    async def nested_services(self):
        return [
            NestedService()
        ]

It is also possible to define dependencies for the requirements. It allows you to wait first services to start before starting the next. Dependencies are defined as names of the other methods wrapped with core_service.requirements() decorator.

from core_service import Service, requirements

class RequiredService(Service):
    pass

class DependentService(Service):
    def __init__(self, required_service):
        self.required_service = required_service

    async def start(self):
        if not self.required_service.running:
            raise RuntimeError("Required service not running)
        await super().start()

class MyService(Service):
    def __init__(self):
        super().__init__()
        self.required_service = RequiredService()

    @requirements()
    async def first_services():
        return [
            self.required_service
        ]

    @requirements('first_services')
    async def second_services():
        return [
            DependentService(self.required_service)
        ]

DependentService will be started only after RequiredService startup complete.