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_condition import EntityConditionEQ
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=EntityConditionEQ(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'),
            context=ImmovableServer.Context(session=session, project_name='Example'),
        )

        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_condition import EntityConditionEQ
from entity_server_connector.entity_order import EntityOrderASC
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=EntityConditionEQ(NetrisRestreamer.c.type_id.name, NETRIS_RESTREAMER_TYPE_ID),
            order_by=EntityOrderASC(NetrisRestreamer.c.id.name),
            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'),
            context=ImmovableServer.Context(session=session, project_name='Example'),
        )

        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 dataclasses import field

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:
    id: int = field(init=False, default=None)
    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 = '') -> ViewScene:
        view_scene = ViewScene(title, description)
        view_scene_id = await self._connector.add_one(view_scene)
        view_scene.id = view_scene_id
        return view_scene


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

        view_scene = await immovable_server.add_view_scene('example')
        print(f'view_scene: {view_scene}')


asyncio.run(main())

This prints:

view_scene: ViewScene(id=15, title='example', description='')

EXAMPLE 4: Add records

import asyncio
from dataclasses import field

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:
    id: int = field(init=False, default=None)
    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_scenes(self, *view_scenes: ViewScene) -> None:
        view_scene_ids = await self._connector.add_list(*view_scenes)
        for view_scene_id, view_scene in zip(view_scene_ids, view_scenes):
            view_scene.id = 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'),
            context=ImmovableServer.Context(session=session, project_name='Example'),
        )

        view_scenes = [ViewScene('example1'), ViewScene('example2')]
        print(f'before add: {view_scenes}')
        await immovable_server.add_view_scenes(*view_scenes)
        print(f'after add: {view_scenes}')


asyncio.run(main())

This prints:

before add: [ViewScene(id=None, title='example1', description=''), ViewScene(id=None, title='example2', description='')]
after add: [ViewScene(id=15, title='example1', description=''), ViewScene(id=16, title='example2', description='')]

EXAMPLE 5: 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_condition import EntityConditionIN


@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=EntityConditionIN(ViewScene.c.id.name, ids),
        )
        return deletion_status


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

        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