Sync client¶

Quickstart¶

Here is a short example of using sync client to interact with backend.

Create client object:

>>> from horizon.client.sync import HorizonClientSync
>>> from horizon.client.auth import LoginPassword
>>> client = HorizonClientSync(
...     base_url="http://some.domain.com/api",
...     auth=LoginPassword(login="me", password="12345"),
... )

Check for credentials and issue access token:

>>> client.authorize()

Create namespace with name “my_namespace”:

>>> from horizon.commons.schemas.v1 import NamespaceCreateRequestV1
>>> created_namespace = client.create_namespace(NamespaceCreateRequestV1(name="my_namespace"))
>>> created_namespace
NamespaceResponseV1(
    id=1,
    name="my_namespace",
    description="",
)

Create HWM with name “my_hwm” in this namespace:

>>> from horizon.commons.schemas.v1 import HWMCreateRequestV1
>>> hwm = HWMCreateRequestV1(
...     namespace_id=created_namespace.id,
...     name="my_hwm",
...     type="column_int",
...     value=123,
... )
>>> created_hwm = client.create_hwm(hwm)
>>> created_hwm
HWMResponseV1(
    id=1,
    namespace_id=1,
    name="my_hwm",
    description="",
    type="column_int",
    value=123,
    entity="",
    expression="",
)

Update HWM with name “my_hwm” in this namespace:

>>> from horizon.commons.schemas.v1 import HWMUpdateRequestV1
>>> hwm_change = HWMUpdateRequestV1(value=234)
>>> updated_hwm = client.update_hwm(created_hwm.id, hwm_change)
>>> updated_hwm
HWMResponseV1(
    id=1,
    namespace_id=1,
    name="my_hwm",
    description="",
    type="column_int",
    value=234,
    entity="",
    expression="",
)

Reference¶

class horizon.client.sync.HorizonClientSync(*, base_url: ~pydantic.networks.AnyHttpUrl, auth: ~horizon.client.auth.base.BaseAuth, session: ~authlib.integrations.requests_client.oauth2_session.OAuth2Session | None = None, retry: ~horizon.client.sync.RetryConfig = <factory>, timeout: ~horizon.client.sync.TimeoutConfig = <factory>)¶

Sync Horizon client implementation, based on authlib and requests.

Parameters:

base_urlstr: URL of Horizon API, e.g. https://some.domain.com/api
authBaseAuth: Authentication class
retryRetryConfig: Configuration for request retries.
timeoutTimeoutConfig: Configuration for request timeouts.
sessionauthlib.integrations.requests_client.OAuth2Session: Custom session object. Inherited from requests.Session, so you can pass custom session options.

Methods

`authorize`()	Fetch and set access token (if required).
`bulk_copy_hwm`(data)	Copy HWMs from one namespace to another.
`bulk_delete_hwm`(namespace_id, hwm_ids)	Bulk delete HWMs.
`create_hwm`(data)	Create new HWM.
`create_namespace`(data)	Create new namespace.
`delete_hwm`(hwm_id)	Delete existing HWM.
`delete_namespace`(namespace_id)	Delete existing namespace.
`get_hwm`(hwm_id)	Get HWM.
`get_namespace`(namespace_id)	Get namespace by name.
`get_namespace_permissions`(namespace_id)	Get permissions for a namespace.
`paginate_hwm`(query)	Get page with HWMs.
`paginate_hwm_history`(query)	Get page with HWM changes history.
`paginate_namespace_history`(query)	Get page with namespace changes history.
`paginate_namespaces`([query])	Get page with namespaces.
`ping`()	Ping Horizon server.
`update_hwm`(hwm_id, changes)	Update existing HWM.
`update_namespace`(namespace_id, changes)	Update existing namespace.
`update_namespace_permissions`(namespace_id, ...)	Update permissions for a namespace.
`whoami`()	Get current user info.

Examples

Using default parameters:

>>> from horizon.client.auth import LoginPassword
>>> from horizon.client.sync import HorizonClientSync
>>> client = HorizonClientSync(
...     base_url="https://some.domain.com/api",
...     auth=LoginPassword(login="me", password="12345"),
... )

Customize retry and timeout:

>>> from horizon.client.auth import LoginPassword
>>> from horizon.client.sync import HorizonClientSync, RetryConfig, TimeoutConfig
>>> client = HorizonClientSync(
...     base_url="https://some.domain.com/api",
...     auth=LoginPassword(login="me", password="12345"),
...     retry=RetryConfig(total=2, backoff_factor=10, status_forcelist=[500, 503]),
...     timeout=TimeoutConfig(request_timeout=3.5),
... )

authorize() → None¶

Fetch and set access token (if required).

Raises:

horizon.commons.exceptions.AuthorizationError: Authorization failed

Examples

>>> client.authorize()

ping() → PingResponse¶

Ping Horizon server.

Examples

>>> client.ping()
PingResponse(status="ok")

whoami() → UserResponseV1WithAdmin | UserResponseV1¶

Get current user info.

Examples

>>> client.whoami()
UserResponseV1(
    id=1,
    username="me",
)

>>> client.whoami()  # for a superadmin user:
UserResponseV1WithAdmin(
    id=1,
    username="admin",
    is_admin=True,
)

paginate_namespaces(query: NamespacePaginateQueryV1 | None = None) → PageResponseV1[NamespaceResponseV1]¶

Get page with namespaces.

Parameters:

queryNamespacePaginateQueryV1: Namespace query parameters

Returns:

PageResponseV1 of NamespaceResponseV1: List of namespaces, limited and filtered by query parameters.

Examples

Get all namespaces:

>>> client.paginate_namespaces()
PageResponseV1[NamespaceResponseV1](
    meta=PageMetaResponseV1(
        page=1,
        pages_count=1,
        total_count=10,
        page_size=20,
        has_next=False,
        has_previous=False,
        next_page=None,
        previous_page=None,
    ),
    items=[NamespaceResponseV1(...), ...],
)

Get all namespaces starting with a page number and page size:

>>> from horizon.commons.schemas.v1 import NamespacePaginateQueryV1
>>> namespace_query = NamespacePaginateQueryV1(page=2, page_size=20)
>>> client.paginate_namespaces(query=namespace_query)
PageResponseV1[NamespaceResponseV1](
    meta=PageMetaResponseV1(
        page=2,
        pages_count=3,
        total_count=50,
        page_size=20,
        has_next=True,
        has_previous=True,
        next_page=3,
        previous_page=1,
    ),
    items=[NamespaceResponseV1(...), ...],
)

Search for namespace with specific name:

>>> from horizon.commons.schemas.v1 import NamespacePaginateQueryV1
>>> namespace_query = NamespacePaginateQueryV1(name="my_namespace")
>>> client.paginate_namespaces(query=namespace_query)
PageResponseV1[NamespaceResponseV1](
    meta=PageMetaResponseV1(
        page=1,
        pages_count=1,
        total_count=1,
        page_size=10,
        has_next=False,
        has_previous=False,
        next_page=None,
        previous_page=None,
    ),
    items=[
        NamespaceResponseV1(name="my_namespace", ...),
    ],
)

get_namespace(namespace_id: int) → NamespaceResponseV1¶

Get namespace by name.

Parameters:

namespace_idint: Namespace name to get

Returns:

NamespaceResponseV1: Namespace

Raises:

EntityNotFoundError: Namespace not found

Examples

>>> client.get_namespace(namespace_id=123)
NamespaceResponseV1(
    id=123,
    name="my_namespace",
    ...
)

create_namespace(data: NamespaceCreateRequestV1) → NamespaceResponseV1¶

Create new namespace.

Parameters:

namespaceNamespaceCreateRequestV1: Namespace to create

Returns:

NamespaceResponseV1: Created namespace

Raises:

EntityAlreadyExistsError: Namespace with the same name already exists

Examples

>>> from horizon.commons.schemas.v1 import NamespaceCreateRequestV1
>>> to_create = NamespaceCreateRequestV1(name="my_namespace")
>>> client.create_namespace(data=to_create)
NamespaceResponseV1(
    id=123,
    name="my_namespace",
    ...
)

update_namespace(namespace_id: int, changes: NamespaceUpdateRequestV1) → NamespaceResponseV1¶

Update existing namespace.

Parameters:

namespace_idint: Namespace name to update
changesNamespaceUpdateRequestV1: Changes to namespace object

Returns:

NamespaceResponseV1: Updated namespace

Raises:

EntityNotFoundError: Namespace not found
EntityAlreadyExistsError: Namespace with the same name already exists
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> from horizon.commons.schemas.v1 import NamespaceUpdateRequestV1
>>> to_update = NamespaceUpdateRequestV1(name="new_namespace_name")
>>> client.update_namespace(namespace_id=123, changes=to_update)
NamespaceResponseV1(
    id=123,
    name="new_namespace_name",
    ...
)

delete_namespace(namespace_id: int) → None¶

Delete existing namespace.

Parameters:

namespace_idint: Namespace name to delete

Raises:

EntityNotFoundError: Namespace not found
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> client.delete_namespace(namespace_id=123)

paginate_namespace_history(query: NamespaceHistoryPaginateQueryV1) → PageResponseV1[NamespaceHistoryResponseV1]¶

Get page with namespace changes history.

Parameters:

queryNamespaceHistoryPaginateQueryV1: Namespace history query parameters

Returns:

PageResponseV1 of NamespaceHistoryResponseV1: List of namespace history items, limited and filtered by query parameters.

Examples

Get all changes of specific namespace:

>>> from horizon.commons.schemas.v1 import NamespacePaginateQueryV1
>>> namespace_query = NamespacePaginateQueryV1(namespace_id=234)
>>> client.paginate_namespace(query=namespace_query)
PageResponseV1[NamespaceHistoryResponseV1](
    meta=PageMetaResponseV1(
        page=1,
        pages_count=2,
        total_count=10,
        page_size=10,
        has_next=True,
        has_previous=False,
        next_page=2,
        previous_page=None,
    ),
    items=[NamespaceHistoryResponseV1(namespace_id=234, ...), ...],
)

Get all changes of specific namespace starting with a page number and page size:

>>> from horizon.commons.schemas.v1 import NamespacePaginateQueryV1
>>> namespace_query = NamespacePaginateQueryV1(namespace_id=234, page=2, page_size=20)
>>> client.paginate_namespace(query=namespace_query)
PageResponseV1[NamespaceHistoryResponseV1](
    meta=PageMetaResponseV1(
        page=2,
        pages_count=3,
        total_count=50,
        page_size=20,
        has_next=True,
        has_previous=True,
        next_page=3,
        previous_page=1,
    ),
    items=[NamespaceHistoryResponseV1(namespace_id=234, ...), ...],
)

paginate_hwm(query: HWMPaginateQueryV1) → PageResponseV1[HWMResponseV1]¶

Get page with HWMs.

Parameters:

queryHWMPaginateQueryV1: HWM query parameters

Returns:

PageResponseV1 of HWMResponseV1: List of HWM, limited and filtered by query parameters.

Examples

Get all HWM in namespace with specific id:

>>> from horizon.commons.schemas.v1 import HWMPaginateQueryV1
>>> hwm_query = HWMPaginateQueryV1(namespace_id=123)
>>> client.paginate_hwm(query=hwm_query)
PageResponseV1[HWMResponseV1](
    meta=PageMetaResponseV1(
        page=1,
        pages_count=2,
        total_count=10,
        page_size=10,
        has_next=True,
        has_previous=False,
        next_page=2,
        previous_page=None,
    ),
    items=[HWMResponseV1(namespace_id=123, ...), ...],
)

Get all HWM in namespace starting with a page number and page size:

>>> from horizon.commons.schemas.v1 import HWMPaginateQueryV1
>>> hwm_query = HWMPaginateQueryV1(namespace_id=123, page=2, page_size=20)
>>> client.paginate_hwm(query=hwm_query)
PageResponseV1[HWMResponseV1](
    meta=PageMetaResponseV1(
        page=2,
        pages_count=3,
        total_count=50,
        page_size=20,
        has_next=True,
        has_previous=True,
        next_page=3,
        previous_page=1,
    ),
    items=[HWMResponseV1(namespace_id=123, ...), ...],
)

Search for HWM with specific namespace and name:

>>> from horizon.commons.schemas.v1 import HWMPaginateQueryV1
>>> hwm_query = HWMPaginateQueryV1(namespace_id=123, name="my_hwm")
>>> client.paginate_hwm(query=hwm_query)
PageResponseV1[HWMResponseV1](
    meta=PageMetaResponseV1(
        page=1,
        pages_count=1,
        total_count=1,
        page_size=10,
        has_next=False,
        has_previous=False,
        next_page=None,
        previous_page=None,
    ),
    items=[
        HWMResponseV1(namespace_id=123, name="my_hwm", ...),
    ],
)

get_hwm(hwm_id: int) → HWMResponseV1¶

Get HWM.

Parameters:

hwm_idint: HWM id to get

Returns:

HWMResponseV1: HWM

Raises:

EntityNotFoundError: HWM not found

Examples

>>> client.get_hwm(hwm_id=234)
HWMResponseV1(
    id=234,
    namespace_id=123,
    name="my_hwm",
    ...
)

create_hwm(data: HWMCreateRequestV1) → HWMResponseV1¶

Create new HWM.

Parameters:

dataHWMCreateRequestV1: HWM data

Returns:

HWMResponseV1: Created HWM

Raises:

EntityNotFoundError: Namespace not found
EntityAlreadyExistsError: HWM with the same name already exists
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> from horizon.commons.schemas.v1 import HWMCreateRequestV1
>>> to_create = HWMCreateRequestV1(
...     namespace_id=123,
...     name="my_hwm",
...     type="column_int",
...     value=5678,
... )
>>> client.create_hwm(data=to_create)
HWMResponseV1(
    namespace_id=123,
    id=234,
    name="my_hwm",
    type="column_int",
    value=5678,
    ...,
)

update_hwm(hwm_id: int, changes: HWMUpdateRequestV1) → HWMResponseV1¶

Update existing HWM.

Parameters:

hwm_idint: HWM id to update
changesHWMUpdateRequestV1: HWM changes

Returns:

HWMResponseV1: Updated HWM

Raises:

EntityNotFoundError: HWM not found
EntityAlreadyExistsError: HWM with the same name already exists
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> from horizon.commons.schemas.v1 import HWMUpdateRequestV1
>>> to_update = HWMUpdateRequestV1(type="column_int", value=5678)
>>> client.update_hwm(hwm_id=234, changes=to_update)
HWMResponseV1(
    namespace_id=123,
    id=234,
    name="my_hwm",
    type="column_int",
    value=5678,
    ...,
)

delete_hwm(hwm_id: int) → None¶

Delete existing HWM.

Parameters:

hwm_idint: HWM id to delete

Raises:

EntityNotFoundError: HWM not found
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> client.delete_hwm(hwm_id=234)

bulk_copy_hwm(data: HWMBulkCopyRequestV1) → HWMListResponseV1¶

Copy HWMs from one namespace to another.

Note

Method ignores HWMs that are not related to provided source namespace, or does not exist.

Parameters:

dataHWMBulkCopyRequestV1: HWM copy data

Returns:

HWMListResponseV1: Copied HWMs

Raises:

EntityNotFoundError: Raised if any of the specified namespaces not found.
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> from horizon.commons.schemas.v1 import HWMBulkCopyRequestV1
>>> copy_data = HWMBulkCopyRequestV1(
...     source_namespace_id=123,
...     target_namespace_id=456,
...     hwm_ids=[1, 2, 3],
...     with_history=True,
... )
>>> copied_hwms = client.bulk_copy_hwm(data=copy_data)
[HWMResponseV1(...), HWMResponseV1(...), HWMResponseV1(...)]

bulk_delete_hwm(namespace_id: int, hwm_ids: List[int]) → None¶

Bulk delete HWMs.

Note

Method ignores HWMs that are not related to provided namespace.

Parameters:

namespace_idint: Namespace ID where the HWMs belong.
hwm_idsList[int]: List of HWM IDs to be deleted.

Raises:

PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> client.bulk_delete_hwm(
...     namespace_id=123,
...     hwm_ids=[234, 345, 456]
... )

get_namespace_permissions(namespace_id: int) → PermissionsResponseV1¶

Get permissions for a namespace.

Parameters:

namespace_idint: The ID of the namespace to get permissions for.

Returns:

PermissionsResponseV1: The permissions of the namespace.

Raises:

EntityNotFoundError: Namespace or provided user not found.
PermissionDeniedError: Permission denied for performing the requested action.

Examples

>>> client.get_namespace_permissions(namespace_id=234)

update_namespace_permissions(namespace_id: int, changes: PermissionsUpdateRequestV1) → PermissionsResponseV1¶

Update permissions for a namespace.

Parameters:

namespace_idint: The ID of the namespace to update permissions for.
changesPermissionsUpdateRequestV1: The changes to apply to the namespace’s permissions.

Returns:

PermissionsResponseV1: Actual permissions of the namespace.

Raises:

EntityNotFoundError: Namespace or provided user not found.
PermissionDeniedError: Permission denied for performing the requested action.
BadRequestError: Bad request with incorrect operating logic.

Examples

>>> from horizon.commons.schemas.v1 import PermissionsUpdateRequestV1, PermissionUpdateRequestItemV1
>>> to_update = PermissionsUpdateRequestV1(
...     permissions=[
...         PermissionUpdateRequestItemV1(username="new_owner", role="OWNER"),
...         PermissionUpdateRequestItemV1(username="add_developer", role="DEVELOPER"),
...         PermissionUpdateRequestItemV1(username="make_read_only", role=None),
...     ]
... )
>>> client.update_namespace_permissions(namespace_id=234, changes=to_update)

paginate_hwm_history(query: HWMHistoryPaginateQueryV1) → PageResponseV1[HWMHistoryResponseV1]¶

Get page with HWM changes history.

Parameters:

queryHWMHistoryPaginateQueryV1: HWM history query parameters

Returns:

PageResponseV1 of HWMHistoryResponseV1: List of HWM history items, limited and filtered by query parameters.

Examples

Get all changes of specific HWM:

>>> from horizon.commons.schemas.v1 import HWMPaginateQueryV1
>>> hwm_query = HWMPaginateQueryV1(hwm_id=234)
>>> client.paginate_hwm(query=hwm_query)
PageResponseV1[HWMHistoryResponseV1](
    meta=PageMetaResponseV1(
        page=1,
        pages_count=2,
        total_count=10,
        page_size=10,
        has_next=True,
        has_previous=False,
        next_page=2,
        previous_page=None,
    ),
    items=[HWMHistoryResponseV1(hwm_id=234, ...), ...],
)

Get all changes of specific HWM starting with a page number and page size:

>>> from horizon.commons.schemas.v1 import HWMPaginateQueryV1
>>> hwm_query = HWMPaginateQueryV1(hwm_id=234, page=2, page_size=20)
>>> client.paginate_hwm(query=hwm_query)
PageResponseV1[HWMHistoryResponseV1](
    meta=PageMetaResponseV1(
        page=2,
        pages_count=3,
        total_count=50,
        page_size=20,
        has_next=True,
        has_previous=True,
        next_page=3,
        previous_page=1,
    ),
    items=[HWMHistoryResponseV1(hwm_id=234, ...), ...],
)

class horizon.client.sync.RetryConfig(*, total: int = 3, backoff_factor: float = 0.1, status_forcelist: List[int] = [502, 503, 504], backoff_jitter: float | None = None)¶

Configuration for request retries in case of network errors or specific status codes. If provided, it customizes the retry behavior for requests made by the client. urllib3 retry documentation.

Parameters:

totalint, default: 3: The maximum number of retry attempts to make.
backoff_factorfloat, default: 0.1: A backoff factor to apply between attempts after the second try.
status_forcelistlist[int], default: [502, 503, 504]: A set of HTTP status codes that we should force a retry on.
backoff_jitterfloat, default: None: A random jitter amount (between 0 and 1) to add to the backoff delay. Helps to avoid “thundering herd” issues by randomizing the delay times between retries.

Note

Requires urllib>2.0

class horizon.client.sync.TimeoutConfig(*, connection_timeout: float = 3, request_timeout: float = 5)¶

Configuration for connection and request timeouts. If provided, it customizes the timeout behavior for requests made by the client. requests timeout documentation.

Parameters:

connection_timeoutfloat, default: 3: The maximum number of seconds to wait for a connection to the server.
request_timeoutfloat, default: 5: The maximum number of seconds to wait for a response from the server.