Skip to content

Rapidata audience manager

RapidataAudienceManager #

RapidataAudienceManager(openapi_service: OpenAPIService)

Handles everything regarding audiences from creation to retrieval.

A manager for creating, retrieving, and searching for audiences. Audiences are groups of annotators that can be recruited based on example tasks and assigned jobs.

Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py 
19
20
def __init__(self, openapi_service: OpenAPIService):
    self._openapi_service = openapi_service

create_audience #

create_audience(
    name: str, filters: list[RapidataFilter] | None = None
) -> RapidataAudience

Create a new audience.

Creates a new audience with the specified name and optional filters.

Parameters:

Name Type Description Default
name str

The name of the audience.

 required
filters list[RapidataFilter]

The list of filters to apply to the audience. Defaults to None (no filters).

 None

Returns:

Name Type Description
RapidataAudience RapidataAudience

The created audience instance.
Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py 
22
23
24
25
26
27
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
def create_audience(
    self,
    name: str,
    filters: list[RapidataFilter] | None = None,
) -> RapidataAudience:
    """Create a new audience.

    Creates a new audience with the specified name and optional filters.

    Args:
        name (str): The name of the audience.
        filters (list[RapidataFilter], optional): The list of filters to apply to the audience. Defaults to None (no filters).

    Returns:
        RapidataAudience: The created audience instance.
    """
    with tracer.start_as_current_span("RapidataAudienceManager.create_audience"):
        from rapidata.rapidata_client.audience.rapidata_audience import (
            RapidataAudience,
        )
        from rapidata.api_client.models.create_audience_endpoint_input import (
            CreateAudienceEndpointInput,
        )

        logger.debug(f"Creating audience: {name}")
        if filters is None:
            filters = []
        response = self._openapi_service.audience.audience_api.audience_post(
            create_audience_endpoint_input=CreateAudienceEndpointInput(
                name=name,
                filters=[filter._to_audience_model() for filter in filters],
            ),
        )
        audience = RapidataAudience(
            id=response.audience_id,
            name=name,
            filters=filters,
            openapi_service=self._openapi_service,
        )
        return audience

get_audience_by_id #

get_audience_by_id(audience_id: str) -> RapidataAudience

Get an audience by its ID.

Parameters:

Name Type Description Default
audience_id str

The unique identifier of the audience.

 required

Returns:

Name Type Description
RapidataAudience RapidataAudience

The audience instance.
Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py 
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
def get_audience_by_id(self, audience_id: str) -> RapidataAudience:
    """Get an audience by its ID.

    Args:
        audience_id (str): The unique identifier of the audience.

    Returns:
        RapidataAudience: The audience instance.
    """
    with tracer.start_as_current_span("RapidataAudienceManager.get_audience_by_id"):
        from rapidata.rapidata_client.filter._backend_filter_mapper import (
            BackendFilterMapper,
        )
        from rapidata.rapidata_client.audience.rapidata_audience import (
            RapidataAudience,
        )

        logger.debug(f"Getting audience by id: {audience_id}")
        response = (
            self._openapi_service.audience.audience_api.audience_audience_id_get(
                audience_id=audience_id,
            )
        )
        return RapidataAudience(
            id=audience_id,
            name=response.name,
            filters=[
                BackendFilterMapper.backend_filter_from_rapidata_filter(filter)
                for filter in response.filters
            ],
            openapi_service=self._openapi_service,
        )

find_audiences #

find_audiences(
    name: str = "", amount: int = 10, page: int = 1
) -> list[RapidataAudience]

Find your audiences by name.

Parameters:

Name Type Description Default
name str

Filter audiences by name (matching audiences will contain this string). Defaults to "" for any audience.

 ''
amount int

The maximum number of audiences to return. Defaults to 10.

 10
page int

The page of audiences to return. Defaults to 1.

 1

Returns:

Type Description
list[RapidataAudience]

list[RapidataAudience]: A list of RapidataAudience instances.
Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py 
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def find_audiences(
    self, name: str = "", amount: int = 10, page: int = 1
) -> list[RapidataAudience]:
    """Find your audiences by name.

    Args:
        name (str, optional): Filter audiences by name (matching audiences will contain this string). Defaults to "" for any audience.
        amount (int, optional): The maximum number of audiences to return. Defaults to 10.
        page (int, optional): The page of audiences to return. Defaults to 1.

    Returns:
        list[RapidataAudience]: A list of RapidataAudience instances.
    """
    with tracer.start_as_current_span("RapidataAudienceManager.find_audiences"):
        from rapidata.rapidata_client.filter._backend_filter_mapper import (
            BackendFilterMapper,
        )
        from rapidata.api_client.models.audiences_get_name_parameter import (
            AudiencesGetNameParameter,
        )
        from rapidata.rapidata_client.audience.rapidata_audience import (
            RapidataAudience,
        )

        logger.debug(f"Finding audiences: {name}, {amount}")
        response = self._openapi_service.audience.audience_api.audiences_get(
            page=page,
            page_size=amount,
            name=AudiencesGetNameParameter(contains=name),
            sort=["-created_at"],
        )
        audiences = []
        for item in response.items:
            audiences.append(
                RapidataAudience(
                    id=item.id,
                    name=item.name,
                    filters=[
                        BackendFilterMapper.backend_filter_from_rapidata_filter(
                            filter
                        )
                        for filter in item.filters
                    ],
                    openapi_service=self._openapi_service,
                )
            )

        return audiences