Skip to content

Async API

AsyncCatalogApi

Bases: Protocol

Source code in backstage_catalog_client/catalog_api/async_api.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class AsyncCatalogApi(Protocol):
    async def get_entities(
        self,
        **kwargs: Unpack[GetEntitiesRequest],
    ) -> GetEntitiesResponse:
        """
        List catalog entities

        Args:
            kwargs: The request object for getting entities. Defaults to None.

        Returns:
            The response object containing the entities.
        """
        ...

    async def query_entities(
        self,
        **kwargs: Unpack[QueryEntitiesKwargs],
    ) -> QueryEntitiesResponse:
        """
        Gets paginated entities from the catalog.'

        Args:
            **kwargs: keyword arguments, represented as a dict.

        Returns:
            The response object containing the entities.

        Examples:
            ```python
            response = await catalog_client.query_entities(
                search_term='A',
                entity_filter=[{"kind": "User"}],
                order_fields={'field': 'metadata.name', 'order': 'asc'},
                limit=20,
            )
            ```
            this will match all entities of type group having a name starting
            with 'A', ordered by name ascending.
            The response will contain a maximum of 20 entities. In case
            more than 20 entities exist, the response will contain a nextCursor
            property that can be used to fetch the next batch of entities.

            ```python
            second_batch_response = await catalog_client
                .query_entities(cursor=response.page_info.nextCursor)
            ```
            second_batch_response will contain the next batch of (maximum) 20 entities,
            together with a prevCursor property, useful to fetch the previous batch.

        """
        ...

    async def get_entities_by_refs(
        self,
        refs: list[str | EntityRef],
        **opts: Unpack[GetEntitiesByRefsOptions],
    ) -> GetEntitiesByRefsResponse:
        """
        Gets a batch of entities, by their entity refs.
        The output list of entities is of the same size and in the same order as
        the requested list of entity refs. Entries that are not found are returned
        as null.
        """
        ...

    async def get_entity_by_ref(
        self,
        ref: str | EntityRef,
        **options: Unpack[CatalogRequestOptions],
    ) -> Entity | None:
        """
        Gets a single entity from your backstage instance by reference (kind, namespace, name).

        Args:
            ref: The reference to the entity to fetch.
            options: The options for the catalog request. Defaults to None.

        Returns:
            The entity if found, otherwise None.
        """

        ...

    async def get_location_by_entity(
        self,
        ref: str | EntityRef,
        **opts: Unpack[CatalogRequestOptions],
    ) -> Location | None:
        """
        Gets a location associated with an entity.

        Args:
            ref: The reference to the entity to fetch.
            **opts: The options for the catalog request.

        Returns:
            the location if found, otherwise None.
        """
        ...

get_entities(**kwargs) async

List catalog entities

Parameters:

Name Type Description Default
kwargs Unpack[GetEntitiesRequest]

The request object for getting entities. Defaults to None.

 {}

Returns:

Type Description
GetEntitiesResponse

The response object containing the entities.
Source code in backstage_catalog_client/catalog_api/async_api.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
async def get_entities(
    self,
    **kwargs: Unpack[GetEntitiesRequest],
) -> GetEntitiesResponse:
    """
    List catalog entities

    Args:
        kwargs: The request object for getting entities. Defaults to None.

    Returns:
        The response object containing the entities.
    """
    ...

get_entities_by_refs(refs, **opts) async

Gets a batch of entities, by their entity refs. The output list of entities is of the same size and in the same order as the requested list of entity refs. Entries that are not found are returned as null.

Source code in backstage_catalog_client/catalog_api/async_api.py 
82
83
84
85
86
87
88
89
90
91
92
93
async def get_entities_by_refs(
    self,
    refs: list[str | EntityRef],
    **opts: Unpack[GetEntitiesByRefsOptions],
) -> GetEntitiesByRefsResponse:
    """
    Gets a batch of entities, by their entity refs.
    The output list of entities is of the same size and in the same order as
    the requested list of entity refs. Entries that are not found are returned
    as null.
    """
    ...

get_entity_by_ref(ref, **options) async

Gets a single entity from your backstage instance by reference (kind, namespace, name).

Parameters:

Name Type Description Default
ref str | EntityRef

The reference to the entity to fetch.

 required
options Unpack[CatalogRequestOptions]

The options for the catalog request. Defaults to None.

 {}

Returns:

Type Description
Entity | None

The entity if found, otherwise None.
Source code in backstage_catalog_client/catalog_api/async_api.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def get_entity_by_ref(
    self,
    ref: str | EntityRef,
    **options: Unpack[CatalogRequestOptions],
) -> Entity | None:
    """
    Gets a single entity from your backstage instance by reference (kind, namespace, name).

    Args:
        ref: The reference to the entity to fetch.
        options: The options for the catalog request. Defaults to None.

    Returns:
        The entity if found, otherwise None.
    """

    ...

get_location_by_entity(ref, **opts) async

Gets a location associated with an entity.

Parameters:

Name Type Description Default
ref str | EntityRef

The reference to the entity to fetch.

 required
**opts Unpack[CatalogRequestOptions]

The options for the catalog request.

 {}

Returns:

Type Description
Location | None

the location if found, otherwise None.
Source code in backstage_catalog_client/catalog_api/async_api.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
async def get_location_by_entity(
    self,
    ref: str | EntityRef,
    **opts: Unpack[CatalogRequestOptions],
) -> Location | None:
    """
    Gets a location associated with an entity.

    Args:
        ref: The reference to the entity to fetch.
        **opts: The options for the catalog request.

    Returns:
        the location if found, otherwise None.
    """
    ...

query_entities(**kwargs) async

Gets paginated entities from the catalog.'

Parameters:

Name Type Description Default
**kwargs Unpack[QueryEntitiesKwargs]

keyword arguments, represented as a dict.

 {}

Returns:

Type Description
QueryEntitiesResponse

The response object containing the entities.

Examples:

response = await catalog_client.query_entities(
    search_term='A',
    entity_filter=[{"kind": "User"}],
    order_fields={'field': 'metadata.name', 'order': 'asc'},
    limit=20,
)
this will match all entities of type group having a name starting with 'A', ordered by name ascending. The response will contain a maximum of 20 entities. In case more than 20 entities exist, the response will contain a nextCursor property that can be used to fetch the next batch of entities.

second_batch_response = await catalog_client
    .query_entities(cursor=response.page_info.nextCursor)
second_batch_response will contain the next batch of (maximum) 20 entities, together with a prevCursor property, useful to fetch the previous batch.

Source code in backstage_catalog_client/catalog_api/async_api.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def query_entities(
    self,
    **kwargs: Unpack[QueryEntitiesKwargs],
) -> QueryEntitiesResponse:
    """
    Gets paginated entities from the catalog.'

    Args:
        **kwargs: keyword arguments, represented as a dict.

    Returns:
        The response object containing the entities.

    Examples:
        ```python
        response = await catalog_client.query_entities(
            search_term='A',
            entity_filter=[{"kind": "User"}],
            order_fields={'field': 'metadata.name', 'order': 'asc'},
            limit=20,
        )
        ```
        this will match all entities of type group having a name starting
        with 'A', ordered by name ascending.
        The response will contain a maximum of 20 entities. In case
        more than 20 entities exist, the response will contain a nextCursor
        property that can be used to fetch the next batch of entities.

        ```python
        second_batch_response = await catalog_client
            .query_entities(cursor=response.page_info.nextCursor)
        ```
        second_batch_response will contain the next batch of (maximum) 20 entities,
        together with a prevCursor property, useful to fetch the previous batch.

    """
    ...