Skip to content

Rapidata flow manager

RapidataFlowManager #

RapidataFlowManager(openapi_service: OpenAPIService)

Handles everything regarding flows from creation to retrieval.

Source code in src/rapidata/rapidata_client/flow/rapidata_flow_manager.py 
16
17
def __init__(self, openapi_service: OpenAPIService):
    self._openapi_service = openapi_service

create_ranking_flow #

create_ranking_flow(
    name: str,
    instruction: str,
    max_response_threshold: int = 100,
    min_response_threshold: int | None = None,
    validation_set_id: str | None = None,
    settings: Sequence[RapidataSetting] | None = None,
    drain_duration: int | None = None,
    serve_timeout: int | None = None,
) -> RapidataRankingFlow

Create a new ranking flow.

Source code in src/rapidata/rapidata_client/flow/rapidata_flow_manager.py 
19
20
21
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
62
63
64
def create_ranking_flow(
    self,
    name: str,
    instruction: str,
    max_response_threshold: int = 100,
    min_response_threshold: int | None = None,
    validation_set_id: str | None = None,
    settings: Sequence[RapidataSetting] | None = None,
    drain_duration: int | None = None,
    serve_timeout: int | None = None,
) -> RapidataRankingFlow:
    """Create a new ranking flow."""
    if min_response_threshold is None:
        min_response_threshold = max_response_threshold

    with tracer.start_as_current_span("RapidataFlowManager.create_ranking_flow"):
        from rapidata.api_client.models.create_flow_endpoint_input import (
            CreateFlowEndpointInput,
        )

        logger.debug("Creating ranking flow: %s", name)

        response = self._openapi_service.flow.ranking_flow_api.flow_ranking_post(
            create_flow_endpoint_input=CreateFlowEndpointInput(
                name=name,
                criteria=instruction,
                validationSetId=validation_set_id,
                minResponses=min_response_threshold,
                maxResponses=max_response_threshold,
                featureFlags=(
                    [setting._to_feature_flag() for setting in settings]
                    if settings
                    else None
                ),
                drainDurationSeconds=drain_duration,
                serveTimeoutSeconds=serve_timeout,
            ),
        )

        logger.debug("Flow created with id: %s", response.flow_id)

        return RapidataRankingFlow(
            id=response.flow_id,
            name=name,
            openapi_service=self._openapi_service,
        )

create_classify_flow #

create_classify_flow(
    name: str,
    instruction: str,
    categories: list[str] | list[tuple[str, str]],
    max_responses_per_datapoint: int = 15,
    min_responses_per_datapoint: int = 10,
    validation_set_id: str | None = None,
    settings: Sequence[RapidataSetting] | None = None,
    drain_duration: int | None = None,
    serve_timeout: int | None = None,
) -> RapidataClassifyFlow

Create a new classify flow.

Source code in src/rapidata/rapidata_client/flow/rapidata_flow_manager.py 
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def create_classify_flow(
    self,
    name: str,
    instruction: str,
    categories: list[str] | list[tuple[str, str]],
    max_responses_per_datapoint: int = 15,
    min_responses_per_datapoint: int = 10,
    validation_set_id: str | None = None,
    settings: Sequence[RapidataSetting] | None = None,
    drain_duration: int | None = None,
    serve_timeout: int | None = None,
) -> RapidataClassifyFlow:
    """Create a new classify flow."""
    category_pairs: list[tuple[str, str]] = [
        (
            (category, category)
            if isinstance(category, str)
            else (category[0], category[1])
        )
        for category in categories
    ]
    if not 2 <= len(category_pairs) <= 8:
        raise ValueError("Categories must contain between 2 and 8 entries.")
    values = [value for _, value in category_pairs]
    if len(set(values)) != len(values):
        raise ValueError("Category values must be unique.")
    if min_responses_per_datapoint < 1:
        raise ValueError("Min responses per datapoint must be at least 1.")
    if max_responses_per_datapoint < min_responses_per_datapoint:
        raise ValueError(
            "Max responses per datapoint must be at least min responses per datapoint."
        )

    with tracer.start_as_current_span("RapidataFlowManager.create_classify_flow"):
        from rapidata.api_client.models.classify_blueprint_category import (
            ClassifyBlueprintCategory,
        )
        from rapidata.api_client.models.create_simple_flow_endpoint_input import (
            CreateSimpleFlowEndpointInput,
        )
        from rapidata.api_client.models.i_flow_rapid_blueprint_classify_blueprint import (
            IFlowRapidBlueprintClassifyBlueprint,
        )

        logger.debug("Creating classify flow: %s", name)

        response = self._openapi_service.flow.simple_flow_api.flow_simple_post(
            create_simple_flow_endpoint_input=CreateSimpleFlowEndpointInput(
                name=name,
                blueprint=IFlowRapidBlueprintClassifyBlueprint(
                    _t="ClassifyBlueprint",
                    title=instruction,
                    categories=[
                        ClassifyBlueprintCategory(label=label, value=value)
                        for label, value in category_pairs
                    ],
                ),
                maxResponses=max_responses_per_datapoint,
                minResponses=min_responses_per_datapoint,
                validationSetId=validation_set_id,
                featureFlags=(
                    [setting._to_feature_flag() for setting in settings]
                    if settings
                    else None
                ),
                drainDurationSeconds=drain_duration,
                serveTimeoutSeconds=serve_timeout,
            ),
        )

        logger.debug("Flow created with id: %s", response.flow_id)

        return RapidataClassifyFlow(
            id=response.flow_id,
            name=name,
            openapi_service=self._openapi_service,
        )

get_flow_by_id #

get_flow_by_id(
    flow_id: str,
) -> RapidataRankingFlow | RapidataClassifyFlow

Get a flow by its ID.

Source code in src/rapidata/rapidata_client/flow/rapidata_flow_manager.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def get_flow_by_id(
    self, flow_id: str
) -> RapidataRankingFlow | RapidataClassifyFlow:
    """Get a flow by its ID."""
    with tracer.start_as_current_span("RapidataFlowManager.get_flow_by_id"):
        logger.debug("Getting flow by id: %s", flow_id)

        response = self._openapi_service.flow.flow_api.flow_flow_id_get(
            flow_id=flow_id,
        )

        flow = response.to_dict()
        if not isinstance(flow, dict):
            raise ValueError(f"Flow '{flow_id}' returned no flow details.")

        flow_class = (
            RapidataRankingFlow
            if flow_type_from_api(flow["_t"]) == "ranking"
            else RapidataClassifyFlow
        )
        return flow_class(
            id=flow["id"],
            name=flow["name"],
            openapi_service=self._openapi_service,
        )

find_flows #

find_flows(
    name: str = "", amount: int = 10, page: int = 1
) -> list[RapidataRankingFlow | RapidataClassifyFlow]

Find your recent flows.

Source code in src/rapidata/rapidata_client/flow/rapidata_flow_manager.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def find_flows(
    self, name: str = "", amount: int = 10, page: int = 1
) -> list[RapidataRankingFlow | RapidataClassifyFlow]:
    """Find your recent flows."""
    with tracer.start_as_current_span("RapidataFlowManager.find_flows"):
        from rapidata.api_client.models.audience_audience_id_jobs_get_job_id_parameter import (
            AudienceAudienceIdJobsGetJobIdParameter,
        )

        logger.debug("Finding flows, amount: %s", amount)

        response = self._openapi_service.flow.flow_api.flow_get(
            page=page,
            page_size=amount,
            sort=["-created_at"],
            name=AudienceAudienceIdJobsGetJobIdParameter(contains=name),
        )

        return [
            (
                RapidataRankingFlow
                if flow_type_from_api(flow.type.value) == "ranking"
                else RapidataClassifyFlow
            )(
                id=flow.id,
                name=flow.name,
                openapi_service=self._openapi_service,
            )
            for flow in response.items
        ]

preheat #

preheat() -> None

Preheat the boost system to reduce latency for upcoming flow items.

Source code in src/rapidata/rapidata_client/flow/rapidata_flow_manager.py 
201
202
203
204
205
def preheat(self) -> None:
    """Preheat the boost system to reduce latency for upcoming flow items."""
    with tracer.start_as_current_span("RapidataFlowManager.preheat"):
        logger.debug("Preheating boost")
        self._openapi_service.campaign.campaign_api.campaign_boost_preheat_post()