# coding: utf-8
"""
LUSID API
FINBOURNE Technology # noqa: E501
Contact: info@finbourne.com
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
"""
import re # noqa: F401
import io
import warnings
from pydantic.v1 import validate_arguments, ValidationError
from typing import overload, Optional, Union, Awaitable
from typing_extensions import Annotated
from datetime import datetime
from pydantic.v1 import Field, StrictStr, conint, conlist, constr, validator
from typing import Dict, List, Optional
from lusid.models.access_metadata_operation import AccessMetadataOperation
from lusid.models.access_metadata_value import AccessMetadataValue
from lusid.models.custom_entity_request import CustomEntityRequest
from lusid.models.custom_entity_response import CustomEntityResponse
from lusid.models.deleted_entity_response import DeletedEntityResponse
from lusid.models.paged_resource_list_of_custom_entity_response import PagedResourceListOfCustomEntityResponse
from lusid.models.resource_list_of_relationship import ResourceListOfRelationship
from lusid.models.upsert_custom_entities_response import UpsertCustomEntitiesResponse
from lusid.models.upsert_custom_entity_access_metadata_request import UpsertCustomEntityAccessMetadataRequest
from lusid.api_client import ApiClient
from lusid.api_response import ApiResponse
from lusid.exceptions import ( # noqa: F401
ApiTypeError,
ApiValueError
)
[docs]
class CustomEntitiesApi:
"""NOTE: This class is auto generated by OpenAPI Generator
Ref: https://openapi-generator.tech
Do not edit the class manually.
"""
def __init__(self, api_client=None) -> None:
if api_client is None:
api_client = ApiClient.get_default()
self.api_client = api_client
@overload
async def delete_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of Custom Entity to remove.")], identifier_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of Custom Entity to remove.")], identifier_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def delete_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of Custom Entity to remove.")], identifier_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], async_req: Optional[bool]=None, **kwargs) -> Union[DeletedEntityResponse, Awaitable[DeletedEntityResponse]]: # noqa: E501
"""[EARLY ACCESS] DeleteCustomEntity: Delete a Custom Entity instance. # noqa: E501
Delete a Custom Entity instance by a specific entity type. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_custom_entity(entity_type, identifier_type, identifier_value, identifier_scope, async_req=True)
>>> result = thread.get()
:param entity_type: The type of Custom Entity to remove. (required)
:type entity_type: str
:param identifier_type: An identifier type attached to the Custom Entity instance. (required)
:type identifier_type: str
:param identifier_value: The identifier value. (required)
:type identifier_value: str
:param identifier_scope: The identifier scope. (required)
:type identifier_scope: str
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _request_timeout: timeout setting for this request.
If one number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: DeletedEntityResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the delete_custom_entity_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.delete_custom_entity_with_http_info(entity_type, identifier_type, identifier_value, identifier_scope, **kwargs) # noqa: E501
[docs]
@validate_arguments
def delete_custom_entity_with_http_info(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of Custom Entity to remove.")], identifier_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] DeleteCustomEntity: Delete a Custom Entity instance. # noqa: E501
Delete a Custom Entity instance by a specific entity type. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.delete_custom_entity_with_http_info(entity_type, identifier_type, identifier_value, identifier_scope, async_req=True)
>>> result = thread.get()
:param entity_type: The type of Custom Entity to remove. (required)
:type entity_type: str
:param identifier_type: An identifier type attached to the Custom Entity instance. (required)
:type identifier_type: str
:param identifier_value: The identifier value. (required)
:type identifier_value: str
:param identifier_scope: The identifier scope. (required)
:type identifier_scope: str
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the authentication
in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(DeletedEntityResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'entity_type',
'identifier_type',
'identifier_value',
'identifier_scope'
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method delete_custom_entity" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['entity_type']:
_path_params['entityType'] = _params['entity_type']
if _params['identifier_type']:
_path_params['identifierType'] = _params['identifier_type']
if _params['identifier_value']:
_path_params['identifierValue'] = _params['identifier_value']
# process the query parameters
_query_params = []
if _params.get('identifier_scope') is not None: # noqa: E501
_query_params.append(('identifierScope', _params['identifier_scope']))
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params = []
_files = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain', 'application/json', 'text/json']) # noqa: E501
# authentication setting
_auth_settings = ['oauth2'] # noqa: E501
_response_types_map = {
'200': "DeletedEntityResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/customentities/{entityType}/{identifierType}/{identifierValue}', 'DELETE',
_path_params,
_query_params,
_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def delete_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], metadata_key : Annotated[constr(strict=True, max_length=100, min_length=1), Field(..., description="Key of the metadata entry to delete.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which to retrieve the Access Metadata.")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective datetime until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' datetime of the Access Metadata.")] = None, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], metadata_key : Annotated[constr(strict=True, max_length=100, min_length=1), Field(..., description="Key of the metadata entry to delete.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which to retrieve the Access Metadata.")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective datetime until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' datetime of the Access Metadata.")] = None, async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
async def get_all_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Access Metadata. Defaults to returning the latest version of the metadata if not specified.")] = None, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
def get_all_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Access Metadata. Defaults to returning the latest version of the metadata if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
async def get_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to retrieve. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], as_at : Annotated[Optional[datetime], Field(description="The AsAt datetime at which to retrieve the Custom Entity instance.")] = None, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the Custom Entity instance. Defaults to the current LUSID system datetime if not specified.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entity in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> CustomEntityResponse: # noqa: E501
...
@overload
def get_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to retrieve. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], as_at : Annotated[Optional[datetime], Field(description="The AsAt datetime at which to retrieve the Custom Entity instance.")] = None, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the Custom Entity instance. Defaults to the current LUSID system datetime if not specified.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entity in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=True, **kwargs) -> CustomEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def get_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to retrieve. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], as_at : Annotated[Optional[datetime], Field(description="The AsAt datetime at which to retrieve the Custom Entity instance.")] = None, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the Custom Entity instance. Defaults to the current LUSID system datetime if not specified.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entity in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[CustomEntityResponse, Awaitable[CustomEntityResponse]]: # noqa: E501
"""[EARLY ACCESS] GetCustomEntity: Get a Custom Entity instance. # noqa: E501
Retrieve a Custom Entity instance by a specific entity type at a point in AsAt time. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_custom_entity(entity_type, identifier_type, identifier_value, identifier_scope, as_at, effective_at, related_entity_property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param entity_type: The type of Custom Entity to retrieve. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions. (required)
:type entity_type: str
:param identifier_type: An identifier type attached to the Custom Entity instance. (required)
:type identifier_type: str
:param identifier_value: The identifier value. (required)
:type identifier_value: str
:param identifier_scope: The identifier scope. (required)
:type identifier_scope: str
:param as_at: The AsAt datetime at which to retrieve the Custom Entity instance.
:type as_at: datetime
:param effective_at: The effective datetime or cut label at which to get the Custom Entity instance. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param related_entity_property_keys: A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.
:type related_entity_property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the entity in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.
:type relationship_definition_ids: List[str]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _request_timeout: timeout setting for this request.
If one number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: CustomEntityResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_custom_entity_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.get_custom_entity_with_http_info(entity_type, identifier_type, identifier_value, identifier_scope, as_at, effective_at, related_entity_property_keys, relationship_definition_ids, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_custom_entity_with_http_info(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to retrieve. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], as_at : Annotated[Optional[datetime], Field(description="The AsAt datetime at which to retrieve the Custom Entity instance.")] = None, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the Custom Entity instance. Defaults to the current LUSID system datetime if not specified.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entity in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetCustomEntity: Get a Custom Entity instance. # noqa: E501
Retrieve a Custom Entity instance by a specific entity type at a point in AsAt time. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_custom_entity_with_http_info(entity_type, identifier_type, identifier_value, identifier_scope, as_at, effective_at, related_entity_property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param entity_type: The type of Custom Entity to retrieve. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions. (required)
:type entity_type: str
:param identifier_type: An identifier type attached to the Custom Entity instance. (required)
:type identifier_type: str
:param identifier_value: The identifier value. (required)
:type identifier_value: str
:param identifier_scope: The identifier scope. (required)
:type identifier_scope: str
:param as_at: The AsAt datetime at which to retrieve the Custom Entity instance.
:type as_at: datetime
:param effective_at: The effective datetime or cut label at which to get the Custom Entity instance. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param related_entity_property_keys: A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.
:type related_entity_property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the entity in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.
:type relationship_definition_ids: List[str]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the authentication
in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(CustomEntityResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'entity_type',
'identifier_type',
'identifier_value',
'identifier_scope',
'as_at',
'effective_at',
'related_entity_property_keys',
'relationship_definition_ids'
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method get_custom_entity" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['entity_type']:
_path_params['entityType'] = _params['entity_type']
if _params['identifier_type']:
_path_params['identifierType'] = _params['identifier_type']
if _params['identifier_value']:
_path_params['identifierValue'] = _params['identifier_value']
# process the query parameters
_query_params = []
if _params.get('identifier_scope') is not None: # noqa: E501
_query_params.append(('identifierScope', _params['identifier_scope']))
if _params.get('as_at') is not None: # noqa: E501
if isinstance(_params['as_at'], datetime):
_query_params.append(('asAt', _params['as_at'].strftime(self.api_client.configuration.datetime_format)))
else:
_query_params.append(('asAt', _params['as_at']))
if _params.get('effective_at') is not None: # noqa: E501
_query_params.append(('effectiveAt', _params['effective_at']))
if _params.get('related_entity_property_keys') is not None: # noqa: E501
_query_params.append(('relatedEntityPropertyKeys', _params['related_entity_property_keys']))
_collection_formats['relatedEntityPropertyKeys'] = 'multi'
if _params.get('relationship_definition_ids') is not None: # noqa: E501
_query_params.append(('relationshipDefinitionIds', _params['relationship_definition_ids']))
_collection_formats['relationshipDefinitionIds'] = 'multi'
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params = []
_files = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain', 'application/json', 'text/json']) # noqa: E501
# authentication setting
_auth_settings = ['oauth2'] # noqa: E501
_response_types_map = {
'200': "CustomEntityResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/customentities/{entityType}/{identifierType}/{identifierValue}', 'GET',
_path_params,
_query_params,
_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def get_custom_entity_access_metadata_by_key(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], metadata_key : Annotated[constr(strict=True, max_length=100, min_length=1), Field(..., description="Key of the metadata entry to retrieve")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Access Metadata. Defaults to returning the latest version of the metadata if not specified.")] = None, **kwargs) -> List[AccessMetadataValue]: # noqa: E501
...
@overload
def get_custom_entity_access_metadata_by_key(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], metadata_key : Annotated[constr(strict=True, max_length=100, min_length=1), Field(..., description="Key of the metadata entry to retrieve")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to get the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Access Metadata. Defaults to returning the latest version of the metadata if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> List[AccessMetadataValue]: # noqa: E501
...
@overload
async def get_custom_entity_relationships(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of entity get relationships for.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to get relationships. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve relationships. Defaults to return the latest LUSID AsAt time if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter relationships. Users should provide null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifiers types (as property keys) used for referencing Persons or Legal Entities. These take the format {domain}/{scope}/{code} e.g. \"Person/CompanyDetails/Role\". They must be from the \"Person\" or \"LegalEntity\" domain. Only identifier types stated will be used to look up relevant entities in relationships. If not applicable, provide an empty array.")] = None, **kwargs) -> ResourceListOfRelationship: # noqa: E501
...
@overload
def get_custom_entity_relationships(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of entity get relationships for.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to get relationships. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve relationships. Defaults to return the latest LUSID AsAt time if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter relationships. Users should provide null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifiers types (as property keys) used for referencing Persons or Legal Entities. These take the format {domain}/{scope}/{code} e.g. \"Person/CompanyDetails/Role\". They must be from the \"Person\" or \"LegalEntity\" domain. Only identifier types stated will be used to look up relevant entities in relationships. If not applicable, provide an empty array.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfRelationship: # noqa: E501
...
[docs]
@validate_arguments
def get_custom_entity_relationships(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of entity get relationships for.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to get relationships. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve relationships. Defaults to return the latest LUSID AsAt time if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter relationships. Users should provide null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifiers types (as property keys) used for referencing Persons or Legal Entities. These take the format {domain}/{scope}/{code} e.g. \"Person/CompanyDetails/Role\". They must be from the \"Person\" or \"LegalEntity\" domain. Only identifier types stated will be used to look up relevant entities in relationships. If not applicable, provide an empty array.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfRelationship, Awaitable[ResourceListOfRelationship]]: # noqa: E501
"""[EARLY ACCESS] GetCustomEntityRelationships: Get Relationships for Custom Entity # noqa: E501
Get relationships for the specified Custom Entity. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_custom_entity_relationships(entity_type, identifier_scope, identifier_type, identifier_value, effective_at, as_at, filter, identifier_types, async_req=True)
>>> result = thread.get()
:param entity_type: The type of entity get relationships for. (required)
:type entity_type: str
:param identifier_scope: The identifier scope. (required)
:type identifier_scope: str
:param identifier_type: An identifier type attached to the Custom Entity. (required)
:type identifier_type: str
:param identifier_value: The identifier value. (required)
:type identifier_value: str
:param effective_at: The effective datetime or cut label at which to get relationships. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to retrieve relationships. Defaults to return the latest LUSID AsAt time if not specified.
:type as_at: datetime
:param filter: Expression to filter relationships. Users should provide null or empty string for this field until further notice.
:type filter: str
:param identifier_types: Identifiers types (as property keys) used for referencing Persons or Legal Entities. These take the format {domain}/{scope}/{code} e.g. \"Person/CompanyDetails/Role\". They must be from the \"Person\" or \"LegalEntity\" domain. Only identifier types stated will be used to look up relevant entities in relationships. If not applicable, provide an empty array.
:type identifier_types: List[str]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _request_timeout: timeout setting for this request.
If one number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: ResourceListOfRelationship
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_custom_entity_relationships_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.get_custom_entity_relationships_with_http_info(entity_type, identifier_scope, identifier_type, identifier_value, effective_at, as_at, filter, identifier_types, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_custom_entity_relationships_with_http_info(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of entity get relationships for.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to get relationships. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve relationships. Defaults to return the latest LUSID AsAt time if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter relationships. Users should provide null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifiers types (as property keys) used for referencing Persons or Legal Entities. These take the format {domain}/{scope}/{code} e.g. \"Person/CompanyDetails/Role\". They must be from the \"Person\" or \"LegalEntity\" domain. Only identifier types stated will be used to look up relevant entities in relationships. If not applicable, provide an empty array.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetCustomEntityRelationships: Get Relationships for Custom Entity # noqa: E501
Get relationships for the specified Custom Entity. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.get_custom_entity_relationships_with_http_info(entity_type, identifier_scope, identifier_type, identifier_value, effective_at, as_at, filter, identifier_types, async_req=True)
>>> result = thread.get()
:param entity_type: The type of entity get relationships for. (required)
:type entity_type: str
:param identifier_scope: The identifier scope. (required)
:type identifier_scope: str
:param identifier_type: An identifier type attached to the Custom Entity. (required)
:type identifier_type: str
:param identifier_value: The identifier value. (required)
:type identifier_value: str
:param effective_at: The effective datetime or cut label at which to get relationships. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to retrieve relationships. Defaults to return the latest LUSID AsAt time if not specified.
:type as_at: datetime
:param filter: Expression to filter relationships. Users should provide null or empty string for this field until further notice.
:type filter: str
:param identifier_types: Identifiers types (as property keys) used for referencing Persons or Legal Entities. These take the format {domain}/{scope}/{code} e.g. \"Person/CompanyDetails/Role\". They must be from the \"Person\" or \"LegalEntity\" domain. Only identifier types stated will be used to look up relevant entities in relationships. If not applicable, provide an empty array.
:type identifier_types: List[str]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the authentication
in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(ResourceListOfRelationship, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'entity_type',
'identifier_scope',
'identifier_type',
'identifier_value',
'effective_at',
'as_at',
'filter',
'identifier_types'
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method get_custom_entity_relationships" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['entity_type']:
_path_params['entityType'] = _params['entity_type']
if _params['identifier_type']:
_path_params['identifierType'] = _params['identifier_type']
if _params['identifier_value']:
_path_params['identifierValue'] = _params['identifier_value']
# process the query parameters
_query_params = []
if _params.get('identifier_scope') is not None: # noqa: E501
_query_params.append(('identifierScope', _params['identifier_scope']))
if _params.get('effective_at') is not None: # noqa: E501
_query_params.append(('effectiveAt', _params['effective_at']))
if _params.get('as_at') is not None: # noqa: E501
if isinstance(_params['as_at'], datetime):
_query_params.append(('asAt', _params['as_at'].strftime(self.api_client.configuration.datetime_format)))
else:
_query_params.append(('asAt', _params['as_at']))
if _params.get('filter') is not None: # noqa: E501
_query_params.append(('filter', _params['filter']))
if _params.get('identifier_types') is not None: # noqa: E501
_query_params.append(('identifierTypes', _params['identifier_types']))
_collection_formats['identifierTypes'] = 'multi'
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params = []
_files = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain', 'application/json', 'text/json']) # noqa: E501
# authentication setting
_auth_settings = ['oauth2'] # noqa: E501
_response_types_map = {
'200': "ResourceListOfRelationship",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/customentities/{entityType}/{identifierType}/{identifierValue}/relationships', 'GET',
_path_params,
_query_params,
_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def list_custom_entities(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to list the entities. Defaults to returning the latest version of each portfolio if not specified.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 100 if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For more information about filtering results, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, sort_by : Annotated[Optional[conlist(StrictStr)], Field(description="A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing entities; this value is returned from the previous call. If a pagination token is provided, the filter, effectiveAt and asAt fields must not have changed since the original request.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entities in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> PagedResourceListOfCustomEntityResponse: # noqa: E501
...
@overload
def list_custom_entities(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to list the entities. Defaults to returning the latest version of each portfolio if not specified.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 100 if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For more information about filtering results, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, sort_by : Annotated[Optional[conlist(StrictStr)], Field(description="A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing entities; this value is returned from the previous call. If a pagination token is provided, the filter, effectiveAt and asAt fields must not have changed since the original request.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entities in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=True, **kwargs) -> PagedResourceListOfCustomEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def list_custom_entities(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to list the entities. Defaults to returning the latest version of each portfolio if not specified.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 100 if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For more information about filtering results, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, sort_by : Annotated[Optional[conlist(StrictStr)], Field(description="A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing entities; this value is returned from the previous call. If a pagination token is provided, the filter, effectiveAt and asAt fields must not have changed since the original request.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entities in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[PagedResourceListOfCustomEntityResponse, Awaitable[PagedResourceListOfCustomEntityResponse]]: # noqa: E501
"""[EARLY ACCESS] ListCustomEntities: List Custom Entities of the specified entityType. # noqa: E501
List all the Custom Entities matching particular criteria. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.list_custom_entities(entity_type, effective_at, as_at, limit, filter, sort_by, page, related_entity_property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param entity_type: The type of Custom Entity to list. (required)
:type entity_type: str
:param effective_at: The effective datetime or cut label at which to list the entities. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to list the entities. Defaults to returning the latest version of each portfolio if not specified.
:type as_at: datetime
:param limit: When paginating, limit the results to this number. Defaults to 100 if not specified.
:type limit: int
:param filter: Expression to filter the results. For more information about filtering results, see https://support.lusid.com/knowledgebase/article/KA-01914.
:type filter: str
:param sort_by: A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\".
:type sort_by: List[str]
:param page: The pagination token to use to continue listing entities; this value is returned from the previous call. If a pagination token is provided, the filter, effectiveAt and asAt fields must not have changed since the original request.
:type page: str
:param related_entity_property_keys: A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.
:type related_entity_property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the entities in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.
:type relationship_definition_ids: List[str]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _request_timeout: timeout setting for this request.
If one number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: PagedResourceListOfCustomEntityResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the list_custom_entities_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.list_custom_entities_with_http_info(entity_type, effective_at, as_at, limit, filter, sort_by, page, related_entity_property_keys, relationship_definition_ids, **kwargs) # noqa: E501
[docs]
@validate_arguments
def list_custom_entities_with_http_info(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of Custom Entity to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the entities. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to list the entities. Defaults to returning the latest version of each portfolio if not specified.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 100 if not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For more information about filtering results, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, sort_by : Annotated[Optional[conlist(StrictStr)], Field(description="A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing entities; this value is returned from the previous call. If a pagination token is provided, the filter, effectiveAt and asAt fields must not have changed since the original request.")] = None, related_entity_property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")] = None, relationship_definition_ids : Annotated[Optional[conlist(StrictStr)], Field(description="A list of relationship definitions that are used to decorate related entities onto the entities in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] ListCustomEntities: List Custom Entities of the specified entityType. # noqa: E501
List all the Custom Entities matching particular criteria. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.list_custom_entities_with_http_info(entity_type, effective_at, as_at, limit, filter, sort_by, page, related_entity_property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param entity_type: The type of Custom Entity to list. (required)
:type entity_type: str
:param effective_at: The effective datetime or cut label at which to list the entities. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to list the entities. Defaults to returning the latest version of each portfolio if not specified.
:type as_at: datetime
:param limit: When paginating, limit the results to this number. Defaults to 100 if not specified.
:type limit: int
:param filter: Expression to filter the results. For more information about filtering results, see https://support.lusid.com/knowledgebase/article/KA-01914.
:type filter: str
:param sort_by: A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\".
:type sort_by: List[str]
:param page: The pagination token to use to continue listing entities; this value is returned from the previous call. If a pagination token is provided, the filter, effectiveAt and asAt fields must not have changed since the original request.
:type page: str
:param related_entity_property_keys: A list of property keys from any domain that supports relationships to decorate onto related entities. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.
:type related_entity_property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the entities in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.
:type relationship_definition_ids: List[str]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the authentication
in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(PagedResourceListOfCustomEntityResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'entity_type',
'effective_at',
'as_at',
'limit',
'filter',
'sort_by',
'page',
'related_entity_property_keys',
'relationship_definition_ids'
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method list_custom_entities" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['entity_type']:
_path_params['entityType'] = _params['entity_type']
# process the query parameters
_query_params = []
if _params.get('effective_at') is not None: # noqa: E501
_query_params.append(('effectiveAt', _params['effective_at']))
if _params.get('as_at') is not None: # noqa: E501
if isinstance(_params['as_at'], datetime):
_query_params.append(('asAt', _params['as_at'].strftime(self.api_client.configuration.datetime_format)))
else:
_query_params.append(('asAt', _params['as_at']))
if _params.get('limit') is not None: # noqa: E501
_query_params.append(('limit', _params['limit']))
if _params.get('filter') is not None: # noqa: E501
_query_params.append(('filter', _params['filter']))
if _params.get('sort_by') is not None: # noqa: E501
_query_params.append(('sortBy', _params['sort_by']))
_collection_formats['sortBy'] = 'multi'
if _params.get('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
if _params.get('related_entity_property_keys') is not None: # noqa: E501
_query_params.append(('relatedEntityPropertyKeys', _params['related_entity_property_keys']))
_collection_formats['relatedEntityPropertyKeys'] = 'multi'
if _params.get('relationship_definition_ids') is not None: # noqa: E501
_query_params.append(('relationshipDefinitionIds', _params['relationship_definition_ids']))
_collection_formats['relationshipDefinitionIds'] = 'multi'
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params = []
_files = {}
# process the body parameter
_body_params = None
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain', 'application/json', 'text/json']) # noqa: E501
# authentication setting
_auth_settings = ['oauth2'] # noqa: E501
_response_types_map = {
'200': "PagedResourceListOfCustomEntityResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/customentities/{entityType}', 'GET',
_path_params,
_query_params,
_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def patch_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], access_metadata_operation : Annotated[conlist(AccessMetadataOperation), Field(..., description="The Json Patch document")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which the Access Metadata will be effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective datetime until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' datetime of the Access Metadata")] = None, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
def patch_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], access_metadata_operation : Annotated[conlist(AccessMetadataOperation), Field(..., description="The Json Patch document")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which the Access Metadata will be effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective datetime until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' datetime of the Access Metadata")] = None, async_req: Optional[bool]=True, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
async def upsert_custom_entities(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], success_mode : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Whether the batch request should fail Atomically or in a Partial fashion - Allowed Values: Atomic, Partial")], request_body : Annotated[Dict[str, CustomEntityRequest], Field(..., description="The payload describing the Custom Entity instances")], **kwargs) -> UpsertCustomEntitiesResponse: # noqa: E501
...
@overload
def upsert_custom_entities(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], success_mode : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Whether the batch request should fail Atomically or in a Partial fashion - Allowed Values: Atomic, Partial")], request_body : Annotated[Dict[str, CustomEntityRequest], Field(..., description="The payload describing the Custom Entity instances")], async_req: Optional[bool]=True, **kwargs) -> UpsertCustomEntitiesResponse: # noqa: E501
...
[docs]
@validate_arguments
def upsert_custom_entities(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], success_mode : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Whether the batch request should fail Atomically or in a Partial fashion - Allowed Values: Atomic, Partial")], request_body : Annotated[Dict[str, CustomEntityRequest], Field(..., description="The payload describing the Custom Entity instances")], async_req: Optional[bool]=None, **kwargs) -> Union[UpsertCustomEntitiesResponse, Awaitable[UpsertCustomEntitiesResponse]]: # noqa: E501
"""[EARLY ACCESS] UpsertCustomEntities: Batch upsert instances of Custom Entities # noqa: E501
Note: If using partial failure modes, then it is important to check the response body for failures as any failures will still return a 200 status code # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upsert_custom_entities(entity_type, success_mode, request_body, async_req=True)
>>> result = thread.get()
:param entity_type: The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions. (required)
:type entity_type: str
:param success_mode: Whether the batch request should fail Atomically or in a Partial fashion - Allowed Values: Atomic, Partial (required)
:type success_mode: str
:param request_body: The payload describing the Custom Entity instances (required)
:type request_body: Dict[str, CustomEntityRequest]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _request_timeout: timeout setting for this request.
If one number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: UpsertCustomEntitiesResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the upsert_custom_entities_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.upsert_custom_entities_with_http_info(entity_type, success_mode, request_body, **kwargs) # noqa: E501
[docs]
@validate_arguments
def upsert_custom_entities_with_http_info(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], success_mode : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Whether the batch request should fail Atomically or in a Partial fashion - Allowed Values: Atomic, Partial")], request_body : Annotated[Dict[str, CustomEntityRequest], Field(..., description="The payload describing the Custom Entity instances")], **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] UpsertCustomEntities: Batch upsert instances of Custom Entities # noqa: E501
Note: If using partial failure modes, then it is important to check the response body for failures as any failures will still return a 200 status code # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upsert_custom_entities_with_http_info(entity_type, success_mode, request_body, async_req=True)
>>> result = thread.get()
:param entity_type: The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions. (required)
:type entity_type: str
:param success_mode: Whether the batch request should fail Atomically or in a Partial fashion - Allowed Values: Atomic, Partial (required)
:type success_mode: str
:param request_body: The payload describing the Custom Entity instances (required)
:type request_body: Dict[str, CustomEntityRequest]
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the authentication
in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(UpsertCustomEntitiesResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'entity_type',
'success_mode',
'request_body'
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method upsert_custom_entities" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['entity_type']:
_path_params['entityType'] = _params['entity_type']
# process the query parameters
_query_params = []
if _params.get('success_mode') is not None: # noqa: E501
_query_params.append(('successMode', _params['success_mode']))
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params = []
_files = {}
# process the body parameter
_body_params = None
if _params['request_body'] is not None:
_body_params = _params['request_body']
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain', 'application/json', 'text/json']) # noqa: E501
# set the HTTP header `Content-Type`
_content_types_list = _params.get('_content_type',
self.api_client.select_header_content_type(
['application/json-patch+json', 'application/json', 'text/json', 'application/*+json']))
if _content_types_list:
_header_params['Content-Type'] = _content_types_list
# authentication setting
_auth_settings = ['oauth2'] # noqa: E501
_response_types_map = {
'200': "UpsertCustomEntitiesResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/customentities/{entityType}/$batchUpsert', 'POST',
_path_params,
_query_params,
_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def upsert_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], custom_entity_request : Annotated[CustomEntityRequest, Field(..., description="The payload describing the Custom Entity instance.")], **kwargs) -> CustomEntityResponse: # noqa: E501
...
@overload
def upsert_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], custom_entity_request : Annotated[CustomEntityRequest, Field(..., description="The payload describing the Custom Entity instance.")], async_req: Optional[bool]=True, **kwargs) -> CustomEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def upsert_custom_entity(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], custom_entity_request : Annotated[CustomEntityRequest, Field(..., description="The payload describing the Custom Entity instance.")], async_req: Optional[bool]=None, **kwargs) -> Union[CustomEntityResponse, Awaitable[CustomEntityResponse]]: # noqa: E501
"""[EARLY ACCESS] UpsertCustomEntity: Upsert a Custom Entity instance # noqa: E501
Insert the Custom Entity if it does not exist or update the Custom Entity with the supplied state if it does exist. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upsert_custom_entity(entity_type, custom_entity_request, async_req=True)
>>> result = thread.get()
:param entity_type: The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions. (required)
:type entity_type: str
:param custom_entity_request: The payload describing the Custom Entity instance. (required)
:type custom_entity_request: CustomEntityRequest
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _request_timeout: timeout setting for this request.
If one number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: CustomEntityResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the upsert_custom_entity_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data" # noqa: E501
raise ValueError(message)
if async_req is not None:
kwargs['async_req'] = async_req
return self.upsert_custom_entity_with_http_info(entity_type, custom_entity_request, **kwargs) # noqa: E501
[docs]
@validate_arguments
def upsert_custom_entity_with_http_info(self, entity_type : Annotated[constr(strict=True, max_length=65, min_length=1), Field(..., description="The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions.")], custom_entity_request : Annotated[CustomEntityRequest, Field(..., description="The payload describing the Custom Entity instance.")], **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] UpsertCustomEntity: Upsert a Custom Entity instance # noqa: E501
Insert the Custom Entity if it does not exist or update the Custom Entity with the supplied state if it does exist. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.upsert_custom_entity_with_http_info(entity_type, custom_entity_request, async_req=True)
>>> result = thread.get()
:param entity_type: The type of the Custom Entity to be created. An entityType can be created using the \"CreateCustomEntityDefinition\" endpoint for CustomEntityDefinitions. (required)
:type entity_type: str
:param custom_entity_request: The payload describing the Custom Entity instance. (required)
:type custom_entity_request: CustomEntityRequest
:param async_req: Whether to execute the request asynchronously.
:type async_req: bool, optional
:param _preload_content: if False, the ApiResponse.data will
be set to none and raw_data will store the
HTTP response body without reading/decoding.
Default is True.
:type _preload_content: bool, optional
:param _return_http_data_only: response data instead of ApiResponse
object with status code, headers, etc
:type _return_http_data_only: bool, optional
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the authentication
in the spec for a single request.
:type _request_auth: dict, optional
:type _content_type: string, optional: force content-type for the request
:return: Returns the result object.
If the method is called asynchronously,
returns the request thread.
:rtype: tuple(CustomEntityResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'entity_type',
'custom_entity_request'
]
_all_params.extend(
[
'async_req',
'_return_http_data_only',
'_preload_content',
'_request_timeout',
'_request_auth',
'_content_type',
'_headers'
]
)
# validate the arguments
for _key, _val in _params['kwargs'].items():
if _key not in _all_params:
raise ApiTypeError(
"Got an unexpected keyword argument '%s'"
" to method upsert_custom_entity" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['entity_type']:
_path_params['entityType'] = _params['entity_type']
# process the query parameters
_query_params = []
# process the header parameters
_header_params = dict(_params.get('_headers', {}))
# process the form parameters
_form_params = []
_files = {}
# process the body parameter
_body_params = None
if _params['custom_entity_request'] is not None:
_body_params = _params['custom_entity_request']
# set the HTTP header `Accept`
_header_params['Accept'] = self.api_client.select_header_accept(
['text/plain', 'application/json', 'text/json']) # noqa: E501
# set the HTTP header `Content-Type`
_content_types_list = _params.get('_content_type',
self.api_client.select_header_content_type(
['application/json-patch+json', 'application/json', 'text/json', 'application/*+json']))
if _content_types_list:
_header_params['Content-Type'] = _content_types_list
# authentication setting
_auth_settings = ['oauth2'] # noqa: E501
_response_types_map = {
'200': "CustomEntityResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/customentities/{entityType}', 'POST',
_path_params,
_query_params,
_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
response_types_map=_response_types_map,
auth_settings=_auth_settings,
async_req=_params.get('async_req'),
_return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501
_preload_content=_params.get('_preload_content', True),
_request_timeout=_params.get('_request_timeout'),
collection_formats=_collection_formats,
_request_auth=_params.get('_request_auth'))
@overload
async def upsert_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], metadata_key : Annotated[constr(strict=True, max_length=100, min_length=1), Field(..., description="Key of the metadata entry to retrieve")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], upsert_custom_entity_access_metadata_request : Annotated[UpsertCustomEntityAccessMetadataRequest, Field(..., description="The Custom Entity Access Metadata entry to upsert")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which the Access Metadata will be effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective datetime until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' datetime of the Access Metadata")] = None, **kwargs) -> List[AccessMetadataValue]: # noqa: E501
...
@overload
def upsert_custom_entity_access_metadata(self, entity_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The type of the Custom Entity.")], identifier_type : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="An identifier type attached to the Custom Entity instance.")], identifier_value : Annotated[constr(strict=True, max_length=1024, min_length=1), Field(..., description="The identifier value.")], metadata_key : Annotated[constr(strict=True, max_length=100, min_length=1), Field(..., description="Key of the metadata entry to retrieve")], identifier_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The identifier scope.")], upsert_custom_entity_access_metadata_request : Annotated[UpsertCustomEntityAccessMetadataRequest, Field(..., description="The Custom Entity Access Metadata entry to upsert")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which the Access Metadata will be effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective datetime until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' datetime of the Access Metadata")] = None, async_req: Optional[bool]=True, **kwargs) -> List[AccessMetadataValue]: # noqa: E501
...