pinecone.data.features.bulk_import.bulk_import_request_factory 

 1from enum import Enum
 2from typing import Optional, TypedDict, Any, Union
 3
 4from pinecone.core.openapi.db_data.models import (
 5    StartImportRequest,
 6    ImportErrorMode as ImportErrorModeClass,
 7)
 8
 9from pinecone.utils import parse_non_empty_args, convert_enum_to_string
10
11
12class ImportErrorMode(Enum):
13    CONTINUE = "CONTINUE"
14    ABORT = "ABORT"
15
16
17class DescribeImportArgs(TypedDict, total=False):
18    id: str
19
20
21class CancelImportArgs(TypedDict, total=False):
22    id: str
23
24
25class BulkImportRequestFactory:
26    @staticmethod
27    def start_import_request(
28        uri: str,
29        integration_id: Optional[str] = None,
30        error_mode: Optional[Union[ImportErrorMode, str]] = "CONTINUE",
31    ) -> StartImportRequest:
32        if error_mode is None:
33            error_mode = "CONTINUE"
34        error_mode_str = convert_enum_to_string(error_mode).lower()
35        valid_error_modes = [mode.value.lower() for mode in ImportErrorMode]
36        if error_mode_str not in valid_error_modes:
37            raise ValueError(
38                f"Invalid error_mode: {error_mode}. Must be one of {valid_error_modes}"
39            )
40
41        args_dict = parse_non_empty_args(
42            [
43                ("uri", uri),
44                ("integration_id", integration_id),
45                ("error_mode", ImportErrorModeClass(on_error=error_mode_str)),
46            ]
47        )
48
49        return StartImportRequest(**args_dict)
50
51    @staticmethod
52    def list_imports_paginated_args(
53        limit: Optional[int] = None, pagination_token: Optional[str] = None, **kwargs
54    ) -> dict[str, Any]:
55        return parse_non_empty_args([("limit", limit), ("pagination_token", pagination_token)])
56
57    @staticmethod
58    def describe_import_args(id: str) -> DescribeImportArgs:
59        if isinstance(id, int):
60            id = str(id)
61        return {"id": id}
62
63    @staticmethod
64    def cancel_import_args(id: str) -> CancelImportArgs:
65        if isinstance(id, int):
66            id = str(id)
67        return {"id": id}
class ImportErrorMode(enum.Enum): 
13class ImportErrorMode(Enum):
14    CONTINUE = "CONTINUE"
15    ABORT = "ABORT"

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

  • attribute access::
>>> Color.RED
<Color.RED: 1>
  • value lookup:
>>> Color(1)
<Color.RED: 1>
  • name lookup:
>>> Color['RED']
<Color.RED: 1>

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes -- see the documentation for details.

CONTINUE = <ImportErrorMode.CONTINUE: 'CONTINUE'>
ABORT = <ImportErrorMode.ABORT: 'ABORT'>
Inherited Members
enum.Enum
name
value
class DescribeImportArgs(typing.TypedDict): 
18class DescribeImportArgs(TypedDict, total=False):
19    id: str
id: str
class CancelImportArgs(typing.TypedDict): 
22class CancelImportArgs(TypedDict, total=False):
23    id: str
id: str
class BulkImportRequestFactory: 
26class BulkImportRequestFactory:
27    @staticmethod
28    def start_import_request(
29        uri: str,
30        integration_id: Optional[str] = None,
31        error_mode: Optional[Union[ImportErrorMode, str]] = "CONTINUE",
32    ) -> StartImportRequest:
33        if error_mode is None:
34            error_mode = "CONTINUE"
35        error_mode_str = convert_enum_to_string(error_mode).lower()
36        valid_error_modes = [mode.value.lower() for mode in ImportErrorMode]
37        if error_mode_str not in valid_error_modes:
38            raise ValueError(
39                f"Invalid error_mode: {error_mode}. Must be one of {valid_error_modes}"
40            )
41
42        args_dict = parse_non_empty_args(
43            [
44                ("uri", uri),
45                ("integration_id", integration_id),
46                ("error_mode", ImportErrorModeClass(on_error=error_mode_str)),
47            ]
48        )
49
50        return StartImportRequest(**args_dict)
51
52    @staticmethod
53    def list_imports_paginated_args(
54        limit: Optional[int] = None, pagination_token: Optional[str] = None, **kwargs
55    ) -> dict[str, Any]:
56        return parse_non_empty_args([("limit", limit), ("pagination_token", pagination_token)])
57
58    @staticmethod
59    def describe_import_args(id: str) -> DescribeImportArgs:
60        if isinstance(id, int):
61            id = str(id)
62        return {"id": id}
63
64    @staticmethod
65    def cancel_import_args(id: str) -> CancelImportArgs:
66        if isinstance(id, int):
67            id = str(id)
68        return {"id": id}
@staticmethod
def start_import_request( uri: str, integration_id: Optional[str] = None, error_mode: Union[ImportErrorMode, str, NoneType] = 'CONTINUE') -> pinecone.core.openapi.db_data.model.start_import_request.StartImportRequest: 
27    @staticmethod
28    def start_import_request(
29        uri: str,
30        integration_id: Optional[str] = None,
31        error_mode: Optional[Union[ImportErrorMode, str]] = "CONTINUE",
32    ) -> StartImportRequest:
33        if error_mode is None:
34            error_mode = "CONTINUE"
35        error_mode_str = convert_enum_to_string(error_mode).lower()
36        valid_error_modes = [mode.value.lower() for mode in ImportErrorMode]
37        if error_mode_str not in valid_error_modes:
38            raise ValueError(
39                f"Invalid error_mode: {error_mode}. Must be one of {valid_error_modes}"
40            )
41
42        args_dict = parse_non_empty_args(
43            [
44                ("uri", uri),
45                ("integration_id", integration_id),
46                ("error_mode", ImportErrorModeClass(on_error=error_mode_str)),
47            ]
48        )
49
50        return StartImportRequest(**args_dict)
@staticmethod
def list_imports_paginated_args( limit: Optional[int] = None, pagination_token: Optional[str] = None, **kwargs) -> dict[str, typing.Any]: 
52    @staticmethod
53    def list_imports_paginated_args(
54        limit: Optional[int] = None, pagination_token: Optional[str] = None, **kwargs
55    ) -> dict[str, Any]:
56        return parse_non_empty_args([("limit", limit), ("pagination_token", pagination_token)])
@staticmethod
def describe_import_args( id: str) -> DescribeImportArgs: 
58    @staticmethod
59    def describe_import_args(id: str) -> DescribeImportArgs:
60        if isinstance(id, int):
61            id = str(id)
62        return {"id": id}
@staticmethod
def cancel_import_args( id: str) -> CancelImportArgs: 
64    @staticmethod
65    def cancel_import_args(id: str) -> CancelImportArgs:
66        if isinstance(id, int):
67            id = str(id)
68        return {"id": id}