Skip to content

api

VolurApiAsyncClient dataclass

A client for interacting with Völur API.

This client is used to interact with the Völur API to upload the data securely and effectively. See the actual methods for more details.

Note

This client was not intended to be used directly, instead use the SDK client from [volur.sdk.client.VolurClient][volur.sdk.client].

Source code in src/volur/api/v1alpha1/client.py 
 14
 15
 16
 17
 18
 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
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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
169
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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
@dataclass
class VolurApiAsyncClient:
    """A client for interacting with Völur API.

    This client is used to interact with the Völur API to upload the data
    securely and effectively. See the actual methods for more details.

    Note:
        This client was not intended to be used directly, instead use the SDK client
        from [volur.sdk.client.VolurClient][volur.sdk.client].
    """

    settings: VolurApiSettings = field(default_factory=VolurApiSettings)

    async def upload_materials_information(
        self: "VolurApiAsyncClient",
        materials: AsyncIterator[material_pb2.Material],
    ) -> Status:
        """Uploads Materials Information to the Völur platform using the Völur
        API.

        This method is using a source to get the materials data and then send
        it to the Völur API. This method is asynchronous and will return a
        status of the operation.

        Args:
            materials: a source of materials data to be uploaded to the Völur
                platform.

        Returns:
            The status of the operation.
        """

        async def generate_requests() -> (
            AsyncIterator[material_pb2.UploadMaterialInformationRequest]
        ):
            try:
                async for material in materials:
                    yield material_pb2.UploadMaterialInformationRequest(
                        material=material,
                    )
            except Exception:
                logger.exception(
                    "error occurred while generating requests",
                )

        try:
            logger.info("start uploading materials data")
            channel = grpc.aio.secure_channel(
                self.settings.address,
                grpc.ssl_channel_credentials(),
            )
            stub = material_pb2_grpc.MaterialInformationServiceStub(channel)
            requests = generate_requests()
            stream = stub.UploadMaterialInformation(
                requests,  # type: ignore[arg-type]
                metadata=(
                    (
                        "authorization",
                        f"Bearer {self.settings.token.get_secret_value()}",
                    ),
                ),
            )
            while True:
                response = await stream.read()  # type: ignore[attr-defined]
                if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                    logger.info("successfully uploaded materials information")
                    break
                if response.HasField("status"):
                    if response.status.code != 0:
                        logger.error(
                            f"error occurred while uploading materials information "
                            f"{response.status.code} {response.status.message}",
                        )
                    else:
                        logger.debug(
                            "successfully uploaded materials information",
                        )
                else:
                    raise ValueError("response from a server does not contain status")
            return Status(code=0)
        except grpc.aio.AioRpcError as rpc_error:
            if rpc_error.code() == grpc.StatusCode.UNAUTHENTICATED:
                logger.error(
                    "used token in invalid,"
                    " please set a valid token using"
                    " `VOLUR_API_TOKEN` environment variable",
                )
            else:
                with logger.contextualize(
                    rpc_error_code=rpc_error.code(),
                    rpc_error_details=rpc_error.details(),
                ):
                    logger.exception(
                        "error occurred while uploading materials information",
                    )
            code: int
            code, _ = rpc_error.code().value  # type: ignore[misc]
            message = _ if (_ := rpc_error.details()) else ""
            return Status(
                code=code,
                message=message,
            )

    async def upload_products_information(
        self: "VolurApiAsyncClient",
        products: AsyncIterator[product_pb2.Product],
    ) -> Status:
        """Uploads Products Information to the Völur platform using the Völur
        API.

        This method is using a source to get the products data and then send
        it to the Völur API. This method is asynchronous and will return a
        status of the operation.

        Args:
            products: an iterable of Product protos to be uploaded via API
                platform.

        Returns:
            The status of the operation.
        """

        async def generate_requests() -> (
            AsyncIterator[product_pb2.UploadProductInformationRequest]
        ):
            try:
                async for product in products:
                    yield product_pb2.UploadProductInformationRequest(
                        product=product,
                    )
            except Exception:
                logger.exception(
                    "error occurred while generating requests",
                )

        try:
            logger.info("start uploading products data")
            channel = grpc.aio.secure_channel(
                self.settings.address,
                grpc.ssl_channel_credentials(),
            )
            stub = product_pb2_grpc.ProductInformationServiceStub(channel)
            requests = generate_requests()
            stream = stub.UploadProductInformation(
                requests,  # type: ignore[arg-type]
                metadata=(
                    (
                        "authorization",
                        f"Bearer {self.settings.token.get_secret_value()}",
                    ),
                ),
            )
            while True:
                response = await stream.read()  # type: ignore[attr-defined]
                if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                    logger.info("successfully uploaded products information")
                    break
                if response.HasField("status"):
                    if response.status.code != 0:
                        logger.error(
                            f"error occurred while uploading materials information "
                            f"{response.status.code} {response.status.message}",
                        )
                    else:
                        logger.debug(
                            "successfully uploaded products information",
                        )
                else:
                    raise ValueError("response from a server does not contain status")
            return Status(code=0)
        except grpc.aio.AioRpcError as rpc_error:
            if rpc_error.code() == grpc.StatusCode.UNAUTHENTICATED:
                logger.error(
                    "used token in invalid,"
                    " please set a valid token using"
                    " `VOLUR_API_TOKEN` environment variable",
                )
            else:
                with logger.contextualize(
                    rpc_error_code=rpc_error.code(),
                    rpc_error_details=rpc_error.details(),
                ):
                    logger.exception(
                        "error occurred while uploading products information",
                    )
            code: int
            code, _ = rpc_error.code().value  # type: ignore[misc]
            message = _ if (_ := rpc_error.details()) else ""
            return Status(
                code=code,
                message=message,
            )

    async def upload_demand_information(
        self: "VolurApiAsyncClient",
        demand: AsyncIterator[demand_pb2.Demand],
    ) -> Status:
        """Uploads Demand Information to the Völur platform using the Völur
        API.

        This method is using a source to get the demand data and then send
        it to the Völur API. This method is asynchronous and will return a
        status of the operation.

        Args:
            demand: a source of demand data to be uploaded to the Völur
                platform.

        Returns:
            The status of the operation.
        """

        async def generate_requests() -> (
            AsyncIterator[demand_pb2.UploadDemandInformationRequest]
        ):
            try:
                async for dem in demand:
                    yield demand_pb2.UploadDemandInformationRequest(
                        demand=dem,
                    )
            except Exception:
                logger.exception(
                    "error occurred while generating requests",
                )

        try:
            logger.info("start uploading demand data")
            channel = grpc.aio.secure_channel(
                self.settings.address,
                grpc.ssl_channel_credentials(),
            )
            stub = demand_pb2_grpc.DemandInformationServiceStub(channel)
            requests = generate_requests()
            stream = stub.UploadDemandInformation(
                requests,  # type: ignore[arg-type]
                metadata=(
                    (
                        "authorization",
                        f"Bearer {self.settings.token.get_secret_value()}",
                    ),
                ),
            )
            while True:
                response = await stream.read()  # type: ignore[attr-defined]
                if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                    logger.info("successfully uploaded demand information")
                    break
                if response.HasField("status"):
                    if response.status.code != 0:
                        logger.error(
                            f"error occurred while uploading demand information "
                            f"{response.status.code} {response.status.message}",
                        )
                    else:
                        logger.debug(
                            "successfully uploaded demand information",
                        )
                else:
                    raise ValueError("response from a server does not contain status")
            return Status(code=0)
        except grpc.aio.AioRpcError as rpc_error:
            if rpc_error.code() == grpc.StatusCode.UNAUTHENTICATED:
                logger.error(
                    "used token in invalid,"
                    "please set a valid token using"
                    "`VOLUR_API_TOKEN` environment variable",
                )
            else:
                with logger.contextualize(
                    rpc_error_code=rpc_error.code(),
                    rpc_error_details=rpc_error.details(),
                ):
                    logger.exception(
                        "error occurred while uploading demand information",
                    )
            code: int
            code, _ = rpc_error.code().value  # type: ignore[misc]
            message = _ if (_ := rpc_error.details()) else ""
            return Status(
                code=code,
                message=message,
            )

upload_demand_information(demand) async

Uploads Demand Information to the Völur platform using the Völur API.

This method is using a source to get the demand data and then send it to the Völur API. This method is asynchronous and will return a status of the operation.

Parameters:

Name Type Description Default
demand AsyncIterator[Demand]

a source of demand data to be uploaded to the Völur platform.

 required

Returns:

Type Description
Status

The status of the operation.
Source code in src/volur/api/v1alpha1/client.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
async def upload_demand_information(
    self: "VolurApiAsyncClient",
    demand: AsyncIterator[demand_pb2.Demand],
) -> Status:
    """Uploads Demand Information to the Völur platform using the Völur
    API.

    This method is using a source to get the demand data and then send
    it to the Völur API. This method is asynchronous and will return a
    status of the operation.

    Args:
        demand: a source of demand data to be uploaded to the Völur
            platform.

    Returns:
        The status of the operation.
    """

    async def generate_requests() -> (
        AsyncIterator[demand_pb2.UploadDemandInformationRequest]
    ):
        try:
            async for dem in demand:
                yield demand_pb2.UploadDemandInformationRequest(
                    demand=dem,
                )
        except Exception:
            logger.exception(
                "error occurred while generating requests",
            )

    try:
        logger.info("start uploading demand data")
        channel = grpc.aio.secure_channel(
            self.settings.address,
            grpc.ssl_channel_credentials(),
        )
        stub = demand_pb2_grpc.DemandInformationServiceStub(channel)
        requests = generate_requests()
        stream = stub.UploadDemandInformation(
            requests,  # type: ignore[arg-type]
            metadata=(
                (
                    "authorization",
                    f"Bearer {self.settings.token.get_secret_value()}",
                ),
            ),
        )
        while True:
            response = await stream.read()  # type: ignore[attr-defined]
            if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                logger.info("successfully uploaded demand information")
                break
            if response.HasField("status"):
                if response.status.code != 0:
                    logger.error(
                        f"error occurred while uploading demand information "
                        f"{response.status.code} {response.status.message}",
                    )
                else:
                    logger.debug(
                        "successfully uploaded demand information",
                    )
            else:
                raise ValueError("response from a server does not contain status")
        return Status(code=0)
    except grpc.aio.AioRpcError as rpc_error:
        if rpc_error.code() == grpc.StatusCode.UNAUTHENTICATED:
            logger.error(
                "used token in invalid,"
                "please set a valid token using"
                "`VOLUR_API_TOKEN` environment variable",
            )
        else:
            with logger.contextualize(
                rpc_error_code=rpc_error.code(),
                rpc_error_details=rpc_error.details(),
            ):
                logger.exception(
                    "error occurred while uploading demand information",
                )
        code: int
        code, _ = rpc_error.code().value  # type: ignore[misc]
        message = _ if (_ := rpc_error.details()) else ""
        return Status(
            code=code,
            message=message,
        )

upload_materials_information(materials) async

Uploads Materials Information to the Völur platform using the Völur API.

This method is using a source to get the materials data and then send it to the Völur API. This method is asynchronous and will return a status of the operation.

Parameters:

Name Type Description Default
materials AsyncIterator[Material]

a source of materials data to be uploaded to the Völur platform.

 required

Returns:

Type Description
Status

The status of the operation.
Source code in src/volur/api/v1alpha1/client.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
async def upload_materials_information(
    self: "VolurApiAsyncClient",
    materials: AsyncIterator[material_pb2.Material],
) -> Status:
    """Uploads Materials Information to the Völur platform using the Völur
    API.

    This method is using a source to get the materials data and then send
    it to the Völur API. This method is asynchronous and will return a
    status of the operation.

    Args:
        materials: a source of materials data to be uploaded to the Völur
            platform.

    Returns:
        The status of the operation.
    """

    async def generate_requests() -> (
        AsyncIterator[material_pb2.UploadMaterialInformationRequest]
    ):
        try:
            async for material in materials:
                yield material_pb2.UploadMaterialInformationRequest(
                    material=material,
                )
        except Exception:
            logger.exception(
                "error occurred while generating requests",
            )

    try:
        logger.info("start uploading materials data")
        channel = grpc.aio.secure_channel(
            self.settings.address,
            grpc.ssl_channel_credentials(),
        )
        stub = material_pb2_grpc.MaterialInformationServiceStub(channel)
        requests = generate_requests()
        stream = stub.UploadMaterialInformation(
            requests,  # type: ignore[arg-type]
            metadata=(
                (
                    "authorization",
                    f"Bearer {self.settings.token.get_secret_value()}",
                ),
            ),
        )
        while True:
            response = await stream.read()  # type: ignore[attr-defined]
            if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                logger.info("successfully uploaded materials information")
                break
            if response.HasField("status"):
                if response.status.code != 0:
                    logger.error(
                        f"error occurred while uploading materials information "
                        f"{response.status.code} {response.status.message}",
                    )
                else:
                    logger.debug(
                        "successfully uploaded materials information",
                    )
            else:
                raise ValueError("response from a server does not contain status")
        return Status(code=0)
    except grpc.aio.AioRpcError as rpc_error:
        if rpc_error.code() == grpc.StatusCode.UNAUTHENTICATED:
            logger.error(
                "used token in invalid,"
                " please set a valid token using"
                " `VOLUR_API_TOKEN` environment variable",
            )
        else:
            with logger.contextualize(
                rpc_error_code=rpc_error.code(),
                rpc_error_details=rpc_error.details(),
            ):
                logger.exception(
                    "error occurred while uploading materials information",
                )
        code: int
        code, _ = rpc_error.code().value  # type: ignore[misc]
        message = _ if (_ := rpc_error.details()) else ""
        return Status(
            code=code,
            message=message,
        )

upload_products_information(products) async

Uploads Products Information to the Völur platform using the Völur API.

This method is using a source to get the products data and then send it to the Völur API. This method is asynchronous and will return a status of the operation.

Parameters:

Name Type Description Default
products AsyncIterator[Product]

an iterable of Product protos to be uploaded via API platform.

 required

Returns:

Type Description
Status

The status of the operation.
Source code in src/volur/api/v1alpha1/client.py 
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
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
169
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
200
201
202
203
204
205
206
async def upload_products_information(
    self: "VolurApiAsyncClient",
    products: AsyncIterator[product_pb2.Product],
) -> Status:
    """Uploads Products Information to the Völur platform using the Völur
    API.

    This method is using a source to get the products data and then send
    it to the Völur API. This method is asynchronous and will return a
    status of the operation.

    Args:
        products: an iterable of Product protos to be uploaded via API
            platform.

    Returns:
        The status of the operation.
    """

    async def generate_requests() -> (
        AsyncIterator[product_pb2.UploadProductInformationRequest]
    ):
        try:
            async for product in products:
                yield product_pb2.UploadProductInformationRequest(
                    product=product,
                )
        except Exception:
            logger.exception(
                "error occurred while generating requests",
            )

    try:
        logger.info("start uploading products data")
        channel = grpc.aio.secure_channel(
            self.settings.address,
            grpc.ssl_channel_credentials(),
        )
        stub = product_pb2_grpc.ProductInformationServiceStub(channel)
        requests = generate_requests()
        stream = stub.UploadProductInformation(
            requests,  # type: ignore[arg-type]
            metadata=(
                (
                    "authorization",
                    f"Bearer {self.settings.token.get_secret_value()}",
                ),
            ),
        )
        while True:
            response = await stream.read()  # type: ignore[attr-defined]
            if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                logger.info("successfully uploaded products information")
                break
            if response.HasField("status"):
                if response.status.code != 0:
                    logger.error(
                        f"error occurred while uploading materials information "
                        f"{response.status.code} {response.status.message}",
                    )
                else:
                    logger.debug(
                        "successfully uploaded products information",
                    )
            else:
                raise ValueError("response from a server does not contain status")
        return Status(code=0)
    except grpc.aio.AioRpcError as rpc_error:
        if rpc_error.code() == grpc.StatusCode.UNAUTHENTICATED:
            logger.error(
                "used token in invalid,"
                " please set a valid token using"
                " `VOLUR_API_TOKEN` environment variable",
            )
        else:
            with logger.contextualize(
                rpc_error_code=rpc_error.code(),
                rpc_error_details=rpc_error.details(),
            ):
                logger.exception(
                    "error occurred while uploading products information",
                )
        code: int
        code, _ = rpc_error.code().value  # type: ignore[misc]
        message = _ if (_ := rpc_error.details()) else ""
        return Status(
            code=code,
            message=message,
        )

VolurApiSettings

Bases: BaseSettings

Settings for Völur API client.

This class is used to obtain required configurations for the client to work properly.

This class requires two environment variables to be set:

  • VOLUR_API_ADDRESS: Address of the Völur API server,
  • VOLUR_API_TOKEN: Token for authenticating to the Völur API server.

Please contact Völur to obtain the endpoint address and the token.

Examples:

example.py
1
settings = VolurApiSettings()
Source code in src/volur/api/v1alpha1/settings.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class VolurApiSettings(BaseSettings):
    """Settings for Völur API client.

    This class is used to obtain required configurations for the client to work
    properly.

    This class requires two environment variables to be set:

    - `VOLUR_API_ADDRESS`: Address of the Völur API server,
    - `VOLUR_API_TOKEN`: Token for authenticating to the Völur API server.

    Please contact Völur to obtain the endpoint address and the token.

    Examples:
        ```python title="example.py" linenums="1"
        settings = VolurApiSettings()
        ```
    """

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        env_prefix="VOLUR_API_",
    )
    address: str
    token: SecretStr
    debug: bool = Field(
        False,
        description="Enable debug mode",
    )