[](https://github.com/infrasonar/python-libservice/actions)
[](https://github.com/infrasonar/python-libservice/releases)
# Python library for building InfraSonar Services
This library is created for building [InfraSonar](https://infrasonar.com) services.
## Environment variable
You might want to implement configuration which applies for all assets in all containers, but still don't want to hard-code this setting in your check.
Think for example about a API URL which is always the same, except maybe for a development environment.
The following environment variable are required for a service to work and are always set by InfraSonar:
Environment variable | Default | Description
-------------------- | ---------------------------- | ----------------
`THINGSDB_HOSTLIST` | `thingsdb:9200` | ThingsDB host list.
`THINGSDB_TOKEN` | _empty_ | Token for authentication **(required)**.
`THINGSDB_SCOPE` | `//data` | Collection scope for data.
`HUB_HOST` | `hub` | Hub host
`HUB_PORT` | `8700` | Hub port
`SLEEP_TIME` | `2` | Sleep time in seconds in each iteration.
`LOG_LEVEL` | `warning` | Log level _(error, warning, info, debug)_.
`LOG_COLORIZED` | `0` | Either 0 (=disabled) or 1 (=enabled).
`LOG_FMT` | `%y%m...` | Default format is `%y%m%d %H:%M:%S`.
`DRY_RUN` | _empty_ | If enabled, result data will be printed to stdout instead of send to the hub.
## Usage
```python
from asyncio import AbstractEventLoop
from typing import Tuple, Optional
from libservice import start, Asset, CheckBase
class MyCheck(CheckBase):
# Use CheckBaseMulti if you want to perform checks for multiple assets
# combined. Sometimes this can be useful as you might be able to combine
# multiple assets in a single request.
key = 'my_check'
@classmethod
async def run(cls, ts: float, asset: Asset) -> Tuple[
Optional[dict], Optional[dict]]:
# Return with the state and optionally an error dict which can be
# created using CheckException(my_error_message).to_dict().
# Alternatively, you can rase a CheckException. The return error is
# especially useful with CheckBaseMulti where only a single asset might
# fail or to return an error together with the result.
# For example:
#
# return state, CheckException('Incomplete result').to_dict()
#
return {
'my_type': [
{'name': 'my_item'}
]
}, None
def start_func(loop: AbstractEventLoop):
pass # optional init function
def close_func(loop: AbstractEventLoop):
pass # optional close function
if __name__ == '__main__':
start(
collector_key='my_server',
version='0.1.0',
checks=(MyCheck, ),
start_func=start_func,
close_func=close_func,
no_count=False) # When True, the check(s) do not count (counter + lastseen)
```
## ASCII item names
InfraSonar requires each item to have a unique _name_ property. The value for _name_ must be a _string_ with ASCII compatible character.
When your _name_ is not guaranteed to be ASCII compatible, the following code replaces the incompatible characters with question marks (`?`):
```python
name = name.encode('ascii', errors='replace').decode()
```
## Check is an asset Id is scheduled
In some cases, the dmarc service as an example, you might want to check if an asset Id is scheduled and ask for the container Id.
As the Asset() instance is only available during the check process, you need to verify this in a different way.
```python
from libservice.serviceroom import service_room
container_id = service_room.get_container_id(asset_id)
if container_id is None:
# missing asset ID
...
```
Raw data
{
"_id": null,
"home_page": "https://github.com/infrasonar/python-libservice",
"name": "libservice",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": "monitoring, infrasonar, service",
"author": "Cesbit",
"author_email": "info@cesbit.com",
"download_url": "https://files.pythonhosted.org/packages/0a/f1/59d92ac513b455c1e518bf0efe2e3237683df5ce720b2753e1f37abdc107/libservice-0.1.15.tar.gz",
"platform": null,
"description": "[](https://github.com/infrasonar/python-libservice/actions)\n[](https://github.com/infrasonar/python-libservice/releases)\n\n# Python library for building InfraSonar Services\n\nThis library is created for building [InfraSonar](https://infrasonar.com) services.\n\n## Environment variable\n\nYou might want to implement configuration which applies for all assets in all containers, but still don't want to hard-code this setting in your check.\nThink for example about a API URL which is always the same, except maybe for a development environment.\n\nThe following environment variable are required for a service to work and are always set by InfraSonar:\n\nEnvironment variable | Default | Description\n-------------------- | ---------------------------- | ----------------\n`THINGSDB_HOSTLIST` | `thingsdb:9200` | ThingsDB host list.\n`THINGSDB_TOKEN` | _empty_ | Token for authentication **(required)**.\n`THINGSDB_SCOPE` | `//data` | Collection scope for data.\n`HUB_HOST` | `hub` | Hub host\n`HUB_PORT` | `8700` | Hub port\n`SLEEP_TIME` | `2` | Sleep time in seconds in each iteration.\n`LOG_LEVEL` | `warning` | Log level _(error, warning, info, debug)_.\n`LOG_COLORIZED` | `0` | Either 0 (=disabled) or 1 (=enabled).\n`LOG_FMT` | `%y%m...` | Default format is `%y%m%d %H:%M:%S`.\n`DRY_RUN` | _empty_ | If enabled, result data will be printed to stdout instead of send to the hub.\n\n## Usage\n\n```python\nfrom asyncio import AbstractEventLoop\nfrom typing import Tuple, Optional\nfrom libservice import start, Asset, CheckBase\n\n\nclass MyCheck(CheckBase):\n # Use CheckBaseMulti if you want to perform checks for multiple assets\n # combined. Sometimes this can be useful as you might be able to combine\n # multiple assets in a single request.\n key = 'my_check'\n\n @classmethod\n async def run(cls, ts: float, asset: Asset) -> Tuple[\n Optional[dict], Optional[dict]]:\n # Return with the state and optionally an error dict which can be\n # created using CheckException(my_error_message).to_dict().\n # Alternatively, you can rase a CheckException. The return error is\n # especially useful with CheckBaseMulti where only a single asset might\n # fail or to return an error together with the result.\n # For example:\n #\n # return state, CheckException('Incomplete result').to_dict()\n #\n return {\n 'my_type': [\n {'name': 'my_item'}\n ]\n }, None\n\n\ndef start_func(loop: AbstractEventLoop):\n pass # optional init function\n\ndef close_func(loop: AbstractEventLoop):\n pass # optional close function\n\n\nif __name__ == '__main__':\n start(\n collector_key='my_server',\n version='0.1.0',\n checks=(MyCheck, ),\n start_func=start_func,\n close_func=close_func,\n no_count=False) # When True, the check(s) do not count (counter + lastseen)\n\n```\n\n## ASCII item names\n\nInfraSonar requires each item to have a unique _name_ property. The value for _name_ must be a _string_ with ASCII compatible character.\nWhen your _name_ is not guaranteed to be ASCII compatible, the following code replaces the incompatible characters with question marks (`?`):\n\n```python\nname = name.encode('ascii', errors='replace').decode()\n```\n\n## Check is an asset Id is scheduled\n\nIn some cases, the dmarc service as an example, you might want to check if an asset Id is scheduled and ask for the container Id.\nAs the Asset() instance is only available during the check process, you need to verify this in a different way.\n\n```python\nfrom libservice.serviceroom import service_room\n\ncontainer_id = service_room.get_container_id(asset_id)\nif container_id is None:\n # missing asset ID\n ...\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "Library for building InfraSonar services",
"version": "0.1.15",
"project_urls": {
"Download": "https://github.com/infrasonar/python-libservice/tarball/v0.1.15",
"Homepage": "https://github.com/infrasonar/python-libservice"
},
"split_keywords": [
"monitoring",
" infrasonar",
" service"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0af159d92ac513b455c1e518bf0efe2e3237683df5ce720b2753e1f37abdc107",
"md5": "381e4f934aa359d9df0aacae184fa37b",
"sha256": "131f7b4d85ceb2243f164675ab6e827cfb5dea7e7cf8cd910ea0b461445d1369"
},
"downloads": -1,
"filename": "libservice-0.1.15.tar.gz",
"has_sig": false,
"md5_digest": "381e4f934aa359d9df0aacae184fa37b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 28021,
"upload_time": "2024-04-30T09:54:20",
"upload_time_iso_8601": "2024-04-30T09:54:20.987657Z",
"url": "https://files.pythonhosted.org/packages/0a/f1/59d92ac513b455c1e518bf0efe2e3237683df5ce720b2753e1f37abdc107/libservice-0.1.15.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-30 09:54:20",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "infrasonar",
"github_project": "python-libservice",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "libservice"
}
JSON
