Quick Start

EXAMPLE 1: Get one record

import asyncio

from aiohttp import ClientSession

from entity_server_connector.entity_dataclass import entity_dataclass
from entity_server_connector.entity_elements import EntityCondition
from entity_server_connector.entity_server_connector import EntityServerConnector


@entity_dataclass
class Restreamer:
    id: int
    hostname: str
    port: int


class ImmovableServer:
    Config = EntityServerConnector.Config
    Context = EntityServerConnector.Context

    def __init__(self, config: Config, context: Context) -> None:
        self._connector = EntityServerConnector(config, context)

    async def get_restreamer_by_id(self, restreamer_id: int) -> Restreamer:
        return await self._connector.get_one(Restreamer, filter_by=EntityCondition(Restreamer.c.id.name, restreamer_id))


async def main() -> None:
    async with ClientSession() as session:
        immovable_server = ImmovableServer(
            config=ImmovableServer.Config(url='http://safe_region.ru/immovable_api', server_name='Example'),
            context=ImmovableServer.Context(session=session),
        )

        required_restreamer_id = 1
        restreamer = await immovable_server.get_restreamer_by_id(restreamer_id=required_restreamer_id)
        print(restreamer)


asyncio.run(main())

This prints:

Restreamer(id=1, hostname='10.194.112.21', port=80)

EXAMPLE 2: Get list of records

import asyncio
from typing import Final

from aiohttp import ClientSession

from entity_server_connector.entity_dataclass import entity_dataclass
from entity_server_connector.entity_elements import EntityCondition
from entity_server_connector.entity_server_connector import EntityServerConnector


NETRIS_RESTREAMER_TYPE_ID: Final[int] = 2


@entity_dataclass(entity_name='restreamer')
class NetrisRestreamer:
    id: int
    type_id: int

    def __post_init__(self) -> None:
        assert self.type_id == NETRIS_RESTREAMER_TYPE_ID


class ImmovableServer:
    Config = EntityServerConnector.Config
    Context = EntityServerConnector.Context

    def __init__(self, config: Config, context: Context) -> None:
        self._connector = EntityServerConnector(config, context)

    async def get_netris_restreamers(self, offset: int = 0, limit: int = 10 ** 6) -> list[NetrisRestreamer]:
        return await self._connector.get_list(
            NetrisRestreamer,
            filter_by=EntityCondition(NetrisRestreamer.c.type_id.name, NETRIS_RESTREAMER_TYPE_ID),
            offset=offset,
            limit=limit,
        )


async def main() -> None:
    async with ClientSession() as session:
        immovable_server = ImmovableServer(
            config=ImmovableServer.Config(url='http://safe_region.ru/immovable_api', server_name='Example'),
            context=ImmovableServer.Context(session=session),
        )

        restreamer = await immovable_server.get_netris_restreamers(limit=3)
        print(restreamer)


asyncio.run(main())

This prints:

[NetrisRestreamer(id=2, type_id=2), NetrisRestreamer(id=3, type_id=2), NetrisRestreamer(id=5, type_id=2)]

EXAMPLE 3: Add record

import asyncio

from aiohttp import ClientSession

from entity_server_connector.entity_dataclass import entity_dataclass
from entity_server_connector.entity_server_connector import EntityServerConnector


@entity_dataclass
class ViewScene:
    title: str
    description: str


class ImmovableServer:
    Config = EntityServerConnector.Config
    Context = EntityServerConnector.Context

    def __init__(self, config: Config, context: Context) -> None:
        self._connector = EntityServerConnector(config, context)

    async def add_view_scene(self, title: str, description: str = '') -> int:
        view_scene_id = await self._connector.add_one(ViewScene(title, description))
        return view_scene_id


async def main() -> None:
    async with ClientSession() as session:
        immovable_server = ImmovableServer(
            config=ImmovableServer.Config(url='http://safe_region.ru/immovable_api', server_name='Example'),
            context=ImmovableServer.Context(session=session),
        )

        view_scene_id = await immovable_server.add_view_scene('test_add')
        print(f'view_scene_id={view_scene_id}')


asyncio.run(main())

This prints:

view_scene_id=15

EXAMPLE 4: Delete records

import asyncio

from aiohttp import ClientSession

from entity_server_connector.entity_dataclass import entity_dataclass
from entity_server_connector.entity_server_connector import EntityServerConnector
from entity_server_connector.entity_elements import EntityCondition, ComparisonOperator


@entity_dataclass
class ViewScene:
    id: int


class ImmovableServer:
    Config = EntityServerConnector.Config
    Context = EntityServerConnector.Context

    def __init__(self, config: Config, context: Context) -> None:
        self._connector = EntityServerConnector(config, context)

    async def delete_view_scenes_by_ids(self, ids: list[int]) -> bool:
        deletion_status = await self._connector.delete(
            ViewScene,
            filter_by=EntityCondition(ViewScene.c.id.name, ids, ComparisonOperator.in_),
        )
        return deletion_status


async def main() -> None:
    async with ClientSession() as session:
        immovable_server = ImmovableServer(
            config=ImmovableServer.Config(url='http://10.10.27.38/immovable_api', server_name='Example'),
            context=ImmovableServer.Context(session=session),
        )

        deletion_status = await immovable_server.delete_view_scenes_by_ids([15, 16])
        print(f'deletion_status={deletion_status}')


asyncio.run(main())

This prints:

deletion_status=True