# 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.aggregated_returns_dispersion_request import AggregatedReturnsDispersionRequest
from lusid.models.aggregated_returns_request import AggregatedReturnsRequest
from lusid.models.aggregated_returns_response import AggregatedReturnsResponse
from lusid.models.composite_breakdown_request import CompositeBreakdownRequest
from lusid.models.composite_breakdown_response import CompositeBreakdownResponse
from lusid.models.composite_dispersion_response import CompositeDispersionResponse
from lusid.models.deleted_entity_response import DeletedEntityResponse
from lusid.models.instrument_event_instruction import InstrumentEventInstruction
from lusid.models.instrument_event_instruction_request import InstrumentEventInstructionRequest
from lusid.models.instrument_event_instructions_response import InstrumentEventInstructionsResponse
from lusid.models.model_property import ModelProperty
from lusid.models.operation import Operation
from lusid.models.paged_resource_list_of_instrument_event_instruction import PagedResourceListOfInstrumentEventInstruction
from lusid.models.performance_return import PerformanceReturn
from lusid.models.portfolio import Portfolio
from lusid.models.portfolio_properties import PortfolioProperties
from lusid.models.resource_list_of_access_metadata_value_of import ResourceListOfAccessMetadataValueOf
from lusid.models.resource_list_of_aggregated_return import ResourceListOfAggregatedReturn
from lusid.models.resource_list_of_performance_return import ResourceListOfPerformanceReturn
from lusid.models.resource_list_of_portfolio import ResourceListOfPortfolio
from lusid.models.resource_list_of_processed_command import ResourceListOfProcessedCommand
from lusid.models.resource_list_of_property import ResourceListOfProperty
from lusid.models.resource_list_of_property_interval import ResourceListOfPropertyInterval
from lusid.models.resource_list_of_relation import ResourceListOfRelation
from lusid.models.resource_list_of_relationship import ResourceListOfRelationship
from lusid.models.update_portfolio_request import UpdatePortfolioRequest
from lusid.models.upsert_portfolio_access_metadata_request import UpsertPortfolioAccessMetadataRequest
from lusid.models.upsert_returns_response import UpsertReturnsResponse
from lusid.api_client import ApiClient
from lusid.api_response import ApiResponse
from lusid.exceptions import ( # noqa: F401
ApiTypeError,
ApiValueError
)
[docs]
class PortfoliosApi:
"""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_instrument_event_instruction(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be deleted.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_instrument_event_instruction(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be deleted.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def delete_instrument_event_instruction(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be deleted.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[DeletedEntityResponse, Awaitable[DeletedEntityResponse]]: # noqa: E501
"""[EARLY ACCESS] DeleteInstrumentEventInstruction: Delete Instrument Event Instruction # noqa: E501
Delete a particular instruction for a particular portfolio # 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_instrument_event_instruction(scope, code, instrument_event_instruction_id, portfolio_effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param instrument_event_instruction_id: The id of the instruction to be deleted. (required)
:type instrument_event_instruction_id: str
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: 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_instrument_event_instruction_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_instrument_event_instruction_with_http_info(scope, code, instrument_event_instruction_id, portfolio_effective_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def delete_instrument_event_instruction_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be deleted.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] DeleteInstrumentEventInstruction: Delete Instrument Event Instruction # noqa: E501
Delete a particular instruction for a particular portfolio # 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_instrument_event_instruction_with_http_info(scope, code, instrument_event_instruction_id, portfolio_effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param instrument_event_instruction_id: The id of the instruction to be deleted. (required)
:type instrument_event_instruction_id: str
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: 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 = [
'scope',
'code',
'instrument_event_instruction_id',
'portfolio_effective_at'
]
_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_instrument_event_instruction" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
if _params['instrument_event_instruction_id']:
_path_params['instrumentEventInstructionId'] = _params['instrument_event_instruction_id']
# process the query parameters
_query_params = []
if _params.get('portfolio_effective_at') is not None: # noqa: E501
_query_params.append(('portfolioEffectiveAt', _params['portfolio_effective_at']))
# 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/portfolios/{scope}/{code}/instrumenteventinstructions/{instrumentEventInstructionId}', '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_key_from_portfolio_access_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Quote Access Metadata Rule to retrieve.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], metadata_key : Annotated[constr(strict=True, max_length=256, min_length=1), Field(..., description="The metadataKey identifying the access metadata entry to delete")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective date to delete at, if this is not supplied, it will delete all data found")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective date until which the delete is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' date of the Access Metadata")] = None, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_key_from_portfolio_access_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Quote Access Metadata Rule to retrieve.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], metadata_key : Annotated[constr(strict=True, max_length=256, min_length=1), Field(..., description="The metadataKey identifying the access metadata entry to delete")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective date to delete at, if this is not supplied, it will delete all data found")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective date until which the delete is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' date of the Access Metadata")] = None, async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
async def delete_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def delete_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], async_req: Optional[bool]=None, **kwargs) -> Union[DeletedEntityResponse, Awaitable[DeletedEntityResponse]]: # noqa: E501
"""DeletePortfolio: Delete portfolio # noqa: E501
Delete a particular portfolio. The deletion will take effect from the portfolio's creation datetime. This means that the portfolio will no longer exist at any effective datetime, as per the asAt datetime of deletion. # 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_portfolio(scope, code, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: 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_portfolio_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_portfolio_with_http_info(scope, code, **kwargs) # noqa: E501
[docs]
@validate_arguments
def delete_portfolio_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], **kwargs) -> ApiResponse: # noqa: E501
"""DeletePortfolio: Delete portfolio # noqa: E501
Delete a particular portfolio. The deletion will take effect from the portfolio's creation datetime. This means that the portfolio will no longer exist at any effective datetime, as per the asAt datetime of deletion. # 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_portfolio_with_http_info(scope, code, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: 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 = [
'scope',
'code'
]
_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_portfolio" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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
# 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/portfolios/{scope}/{code}', '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_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_keys : Annotated[conlist(StrictStr), Field(..., description="The property keys of the properties to delete. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. Each property must be from the 'Portfolio' domain.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to delete time-variant properties from. The property must exist at the specified 'effectiveAt' datetime. If the 'effectiveAt' is not provided or is before the time-variant property exists then a failure is returned. Do not specify this parameter if any of the properties to delete are perpetual.")] = None, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_keys : Annotated[conlist(StrictStr), Field(..., description="The property keys of the properties to delete. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. Each property must be from the 'Portfolio' domain.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to delete time-variant properties from. The property must exist at the specified 'effectiveAt' datetime. If the 'effectiveAt' is not provided or is before the time-variant property exists then a failure is returned. Do not specify this parameter if any of the properties to delete are perpetual.")] = None, async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def delete_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_keys : Annotated[conlist(StrictStr), Field(..., description="The property keys of the properties to delete. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. Each property must be from the 'Portfolio' domain.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to delete time-variant properties from. The property must exist at the specified 'effectiveAt' datetime. If the 'effectiveAt' is not provided or is before the time-variant property exists then a failure is returned. Do not specify this parameter if any of the properties to delete are perpetual.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[DeletedEntityResponse, Awaitable[DeletedEntityResponse]]: # noqa: E501
"""DeletePortfolioProperties: Delete portfolio properties # noqa: E501
Delete one or more properties from a particular portfolio. If the properties are time-variant then an effective datetime from which to delete properties must be specified. If the properties are perpetual then it is invalid to specify an effective datetime for deletion. # 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_portfolio_properties(scope, code, property_keys, effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param property_keys: The property keys of the properties to delete. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. Each property must be from the 'Portfolio' domain. (required)
:type property_keys: List[str]
:param effective_at: The effective datetime or cut label at which to delete time-variant properties from. The property must exist at the specified 'effectiveAt' datetime. If the 'effectiveAt' is not provided or is before the time-variant property exists then a failure is returned. Do not specify this parameter if any of the properties to delete are perpetual.
:type effective_at: 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_portfolio_properties_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_portfolio_properties_with_http_info(scope, code, property_keys, effective_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def delete_portfolio_properties_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_keys : Annotated[conlist(StrictStr), Field(..., description="The property keys of the properties to delete. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. Each property must be from the 'Portfolio' domain.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to delete time-variant properties from. The property must exist at the specified 'effectiveAt' datetime. If the 'effectiveAt' is not provided or is before the time-variant property exists then a failure is returned. Do not specify this parameter if any of the properties to delete are perpetual.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""DeletePortfolioProperties: Delete portfolio properties # noqa: E501
Delete one or more properties from a particular portfolio. If the properties are time-variant then an effective datetime from which to delete properties must be specified. If the properties are perpetual then it is invalid to specify an effective datetime for deletion. # 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_portfolio_properties_with_http_info(scope, code, property_keys, effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param property_keys: The property keys of the properties to delete. These must take the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. Each property must be from the 'Portfolio' domain. (required)
:type property_keys: List[str]
:param effective_at: The effective datetime or cut label at which to delete time-variant properties from. The property must exist at the specified 'effectiveAt' datetime. If the 'effectiveAt' is not provided or is before the time-variant property exists then a failure is returned. Do not specify this parameter if any of the properties to delete are perpetual.
:type effective_at: 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 = [
'scope',
'code',
'property_keys',
'effective_at'
]
_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_portfolio_properties" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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('property_keys') is not None: # noqa: E501
_query_params.append(('propertyKeys', _params['property_keys']))
_collection_formats['propertyKeys'] = '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': "DeletedEntityResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/properties', '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_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[StrictStr, Field(..., description="The start date from which to delete the Returns.")], to_effective_at : Annotated[StrictStr, Field(..., description="The end date from which to delete the Returns.")], period : Annotated[Optional[StrictStr], Field(description="The Period (Daily or Monthly) of the Returns to be deleted. Defaults to Daily.")] = None, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
@overload
def delete_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[StrictStr, Field(..., description="The start date from which to delete the Returns.")], to_effective_at : Annotated[StrictStr, Field(..., description="The end date from which to delete the Returns.")], period : Annotated[Optional[StrictStr], Field(description="The Period (Daily or Monthly) of the Returns to be deleted. Defaults to Daily.")] = None, async_req: Optional[bool]=True, **kwargs) -> DeletedEntityResponse: # noqa: E501
...
[docs]
@validate_arguments
def delete_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[StrictStr, Field(..., description="The start date from which to delete the Returns.")], to_effective_at : Annotated[StrictStr, Field(..., description="The end date from which to delete the Returns.")], period : Annotated[Optional[StrictStr], Field(description="The Period (Daily or Monthly) of the Returns to be deleted. Defaults to Daily.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[DeletedEntityResponse, Awaitable[DeletedEntityResponse]]: # noqa: E501
"""[EARLY ACCESS] DeletePortfolioReturns: Delete Returns # noqa: E501
Cancel one or more Returns which exist into the specified portfolio. # 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_portfolio_returns(scope, code, return_scope, return_code, from_effective_at, to_effective_at, period, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param from_effective_at: The start date from which to delete the Returns. (required)
:type from_effective_at: str
:param to_effective_at: The end date from which to delete the Returns. (required)
:type to_effective_at: str
:param period: The Period (Daily or Monthly) of the Returns to be deleted. Defaults to Daily.
:type period: 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_portfolio_returns_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_portfolio_returns_with_http_info(scope, code, return_scope, return_code, from_effective_at, to_effective_at, period, **kwargs) # noqa: E501
[docs]
@validate_arguments
def delete_portfolio_returns_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[StrictStr, Field(..., description="The start date from which to delete the Returns.")], to_effective_at : Annotated[StrictStr, Field(..., description="The end date from which to delete the Returns.")], period : Annotated[Optional[StrictStr], Field(description="The Period (Daily or Monthly) of the Returns to be deleted. Defaults to Daily.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] DeletePortfolioReturns: Delete Returns # noqa: E501
Cancel one or more Returns which exist into the specified portfolio. # 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_portfolio_returns_with_http_info(scope, code, return_scope, return_code, from_effective_at, to_effective_at, period, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param from_effective_at: The start date from which to delete the Returns. (required)
:type from_effective_at: str
:param to_effective_at: The end date from which to delete the Returns. (required)
:type to_effective_at: str
:param period: The Period (Daily or Monthly) of the Returns to be deleted. Defaults to Daily.
:type period: 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 = [
'scope',
'code',
'return_scope',
'return_code',
'from_effective_at',
'to_effective_at',
'period'
]
_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_portfolio_returns" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
if _params['return_scope']:
_path_params['returnScope'] = _params['return_scope']
if _params['return_code']:
_path_params['returnCode'] = _params['return_code']
# process the query parameters
_query_params = []
if _params.get('from_effective_at') is not None: # noqa: E501
_query_params.append(('fromEffectiveAt', _params['from_effective_at']))
if _params.get('to_effective_at') is not None: # noqa: E501
_query_params.append(('toEffectiveAt', _params['to_effective_at']))
if _params.get('period') is not None: # noqa: E501
_query_params.append(('period', _params['period']))
# 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/portfolios/{scope}/{code}/returns/{returnScope}/{returnCode}/$delete', '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 get_aggregated_returns_dispersion_metrics(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_dispersion_request : Annotated[AggregatedReturnsDispersionRequest, Field(..., description="The request used in the AggregatedReturnsDispersionMetric.")], as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> CompositeDispersionResponse: # noqa: E501
...
@overload
def get_aggregated_returns_dispersion_metrics(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_dispersion_request : Annotated[AggregatedReturnsDispersionRequest, Field(..., description="The request used in the AggregatedReturnsDispersionMetric.")], as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=True, **kwargs) -> CompositeDispersionResponse: # noqa: E501
...
[docs]
@validate_arguments
def get_aggregated_returns_dispersion_metrics(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_dispersion_request : Annotated[AggregatedReturnsDispersionRequest, Field(..., description="The request used in the AggregatedReturnsDispersionMetric.")], as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[CompositeDispersionResponse, Awaitable[CompositeDispersionResponse]]: # noqa: E501
"""[EARLY ACCESS] GetAggregatedReturnsDispersionMetrics: Get the Aggregated Returns Dispersion metric # noqa: E501
Calculate the dispersion metric with the Aggregate Returns which are on the specified portfolio. This works only for composites which have at least 6 constituents for a full year in. # 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_aggregated_returns_dispersion_metrics(scope, code, aggregated_returns_dispersion_request, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param aggregated_returns_dispersion_request: The request used in the AggregatedReturnsDispersionMetric. (required)
:type aggregated_returns_dispersion_request: AggregatedReturnsDispersionRequest
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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: CompositeDispersionResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_aggregated_returns_dispersion_metrics_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_aggregated_returns_dispersion_metrics_with_http_info(scope, code, aggregated_returns_dispersion_request, as_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_aggregated_returns_dispersion_metrics_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_dispersion_request : Annotated[AggregatedReturnsDispersionRequest, Field(..., description="The request used in the AggregatedReturnsDispersionMetric.")], as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetAggregatedReturnsDispersionMetrics: Get the Aggregated Returns Dispersion metric # noqa: E501
Calculate the dispersion metric with the Aggregate Returns which are on the specified portfolio. This works only for composites which have at least 6 constituents for a full year in. # 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_aggregated_returns_dispersion_metrics_with_http_info(scope, code, aggregated_returns_dispersion_request, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param aggregated_returns_dispersion_request: The request used in the AggregatedReturnsDispersionMetric. (required)
:type aggregated_returns_dispersion_request: AggregatedReturnsDispersionRequest
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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(CompositeDispersionResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'aggregated_returns_dispersion_request',
'as_at'
]
_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_aggregated_returns_dispersion_metrics" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
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']))
# 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['aggregated_returns_dispersion_request'] is not None:
_body_params = _params['aggregated_returns_dispersion_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': "CompositeDispersionResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/returns/dispersion/$aggregated', '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 get_composite_breakdown(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], composite_breakdown_request : Annotated[CompositeBreakdownRequest, Field(..., description="The request used in the GetCompositeBreakdown.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> CompositeBreakdownResponse: # noqa: E501
...
@overload
def get_composite_breakdown(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], composite_breakdown_request : Annotated[CompositeBreakdownRequest, Field(..., description="The request used in the GetCompositeBreakdown.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=True, **kwargs) -> CompositeBreakdownResponse: # noqa: E501
...
[docs]
@validate_arguments
def get_composite_breakdown(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], composite_breakdown_request : Annotated[CompositeBreakdownRequest, Field(..., description="The request used in the GetCompositeBreakdown.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[CompositeBreakdownResponse, Awaitable[CompositeBreakdownResponse]]: # noqa: E501
"""[EARLY ACCESS] GetCompositeBreakdown: Get the Composite Breakdown on how the composite Returns are calculated # noqa: E501
Calculate the Composite Returns and return this with the constituents which are included in this calculation # 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_composite_breakdown(scope, code, composite_breakdown_request, from_effective_at, to_effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param composite_breakdown_request: The request used in the GetCompositeBreakdown. (required)
:type composite_breakdown_request: CompositeBreakdownRequest
:param from_effective_at: The start date from which to calculate the Returns.
:type from_effective_at: str
:param to_effective_at: The end date for which to calculate the Returns.
:type to_effective_at: str
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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: CompositeBreakdownResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_composite_breakdown_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_composite_breakdown_with_http_info(scope, code, composite_breakdown_request, from_effective_at, to_effective_at, as_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_composite_breakdown_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], composite_breakdown_request : Annotated[CompositeBreakdownRequest, Field(..., description="The request used in the GetCompositeBreakdown.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetCompositeBreakdown: Get the Composite Breakdown on how the composite Returns are calculated # noqa: E501
Calculate the Composite Returns and return this with the constituents which are included in this calculation # 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_composite_breakdown_with_http_info(scope, code, composite_breakdown_request, from_effective_at, to_effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param composite_breakdown_request: The request used in the GetCompositeBreakdown. (required)
:type composite_breakdown_request: CompositeBreakdownRequest
:param from_effective_at: The start date from which to calculate the Returns.
:type from_effective_at: str
:param to_effective_at: The end date for which to calculate the Returns.
:type to_effective_at: str
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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(CompositeBreakdownResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'composite_breakdown_request',
'from_effective_at',
'to_effective_at',
'as_at'
]
_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_composite_breakdown" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('from_effective_at') is not None: # noqa: E501
_query_params.append(('fromEffectiveAt', _params['from_effective_at']))
if _params.get('to_effective_at') is not None: # noqa: E501
_query_params.append(('toEffectiveAt', _params['to_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']))
# 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['composite_breakdown_request'] is not None:
_body_params = _params['composite_breakdown_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': "CompositeBreakdownResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/returns/breakdown', '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 get_instrument_event_instruction(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be retrieved.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instruction. Defaults to return the latest version of the instruction if not specified.")] = None, **kwargs) -> InstrumentEventInstruction: # noqa: E501
...
@overload
def get_instrument_event_instruction(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be retrieved.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instruction. Defaults to return the latest version of the instruction if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> InstrumentEventInstruction: # noqa: E501
...
[docs]
@validate_arguments
def get_instrument_event_instruction(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be retrieved.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instruction. Defaults to return the latest version of the instruction if not specified.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[InstrumentEventInstruction, Awaitable[InstrumentEventInstruction]]: # noqa: E501
"""[EARLY ACCESS] GetInstrumentEventInstruction: Get Instrument Event Instruction # noqa: E501
Get a particular instruction for a particular portfolio # 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_instrument_event_instruction(scope, code, instrument_event_instruction_id, portfolio_effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param instrument_event_instruction_id: The id of the instruction to be retrieved. (required)
:type instrument_event_instruction_id: str
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: str
:param as_at: The asAt datetime at which to retrieve the instruction. Defaults to return the latest version of the instruction if not specified.
:type as_at: datetime
: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: InstrumentEventInstruction
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_instrument_event_instruction_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_instrument_event_instruction_with_http_info(scope, code, instrument_event_instruction_id, portfolio_effective_at, as_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_instrument_event_instruction_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], instrument_event_instruction_id : Annotated[StrictStr, Field(..., description="The id of the instruction to be retrieved.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instruction. Defaults to return the latest version of the instruction if not specified.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetInstrumentEventInstruction: Get Instrument Event Instruction # noqa: E501
Get a particular instruction for a particular portfolio # 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_instrument_event_instruction_with_http_info(scope, code, instrument_event_instruction_id, portfolio_effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param instrument_event_instruction_id: The id of the instruction to be retrieved. (required)
:type instrument_event_instruction_id: str
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: str
:param as_at: The asAt datetime at which to retrieve the instruction. Defaults to return the latest version of the instruction if not specified.
:type as_at: datetime
: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(InstrumentEventInstruction, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'instrument_event_instruction_id',
'portfolio_effective_at',
'as_at'
]
_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_instrument_event_instruction" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
if _params['instrument_event_instruction_id']:
_path_params['instrumentEventInstructionId'] = _params['instrument_event_instruction_id']
# process the query parameters
_query_params = []
if _params.get('portfolio_effective_at') is not None: # noqa: E501
_query_params.append(('portfolioEffectiveAt', _params['portfolio_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']))
# 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': "InstrumentEventInstruction",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/instrumenteventinstructions/{instrumentEventInstructionId}', '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_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve the portfolio definition. 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 portfolio definition. Defaults to returning the latest version of the portfolio definition if not specified.")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto the portfolio, or 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 portfolio in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> Portfolio: # noqa: E501
...
@overload
def get_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve the portfolio definition. 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 portfolio definition. Defaults to returning the latest version of the portfolio definition if not specified.")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto the portfolio, or 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 portfolio in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=True, **kwargs) -> Portfolio: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve the portfolio definition. 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 portfolio definition. Defaults to returning the latest version of the portfolio definition if not specified.")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto the portfolio, or 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 portfolio in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[Portfolio, Awaitable[Portfolio]]: # noqa: E501
"""GetPortfolio: Get portfolio # noqa: E501
Retrieve the definition of a particular portfolio. # 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_portfolio(scope, code, effective_at, as_at, property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to retrieve the portfolio definition. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to retrieve the portfolio definition. Defaults to returning the latest version of the portfolio definition if not specified.
:type as_at: datetime
:param property_keys: A list of property keys from the 'Portfolio' domain to decorate onto the portfolio, or 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 property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the portfolio 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: Portfolio
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_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_portfolio_with_http_info(scope, code, effective_at, as_at, property_keys, relationship_definition_ids, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve the portfolio definition. 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 portfolio definition. Defaults to returning the latest version of the portfolio definition if not specified.")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto the portfolio, or 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 portfolio in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""GetPortfolio: Get portfolio # noqa: E501
Retrieve the definition of a particular portfolio. # 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_portfolio_with_http_info(scope, code, effective_at, as_at, property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to retrieve the portfolio definition. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to retrieve the portfolio definition. Defaults to returning the latest version of the portfolio definition if not specified.
:type as_at: datetime
:param property_keys: A list of property keys from the 'Portfolio' domain to decorate onto the portfolio, or 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 property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the portfolio 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(Portfolio, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'effective_at',
'as_at',
'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_portfolio" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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('property_keys') is not None: # noqa: E501
_query_params.append(('propertyKeys', _params['property_keys']))
_collection_formats['propertyKeys'] = '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': "Portfolio",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}', '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_portfolio_aggregate_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], recipe_id_scope : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Scope for getting the fx rates")] = None, recipe_id_code : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Code for getting the fx rates")] = None, from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, composite_method : Annotated[Optional[StrictStr], Field(description="The method used to calculate the Portfolio performance: Equal/Asset.")] = None, period : Annotated[Optional[StrictStr], Field(description="The type of the returns used to calculate the aggregation result: Daily/Monthly.")] = None, output_frequency : Annotated[Optional[StrictStr], Field(description="The type of calculated output: Daily/Weekly/Monthly/Quarterly/Half-Yearly/Yearly.")] = None, metrics : Annotated[Optional[conlist(StrictStr)], Field(description="Determines what type of returns should be calculated, see https://support.lusid.com/knowledgebase/article/KA-01675/en-us for a list of available metrics.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, alternative_inc_date : Annotated[Optional[StrictStr], Field(description="The date from which to consider the Returns on the Portfolio, if this is different from the date when Returns begin. Can be a date string or Portfolio property.")] = None, **kwargs) -> ResourceListOfAggregatedReturn: # noqa: E501
...
@overload
def get_portfolio_aggregate_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], recipe_id_scope : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Scope for getting the fx rates")] = None, recipe_id_code : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Code for getting the fx rates")] = None, from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, composite_method : Annotated[Optional[StrictStr], Field(description="The method used to calculate the Portfolio performance: Equal/Asset.")] = None, period : Annotated[Optional[StrictStr], Field(description="The type of the returns used to calculate the aggregation result: Daily/Monthly.")] = None, output_frequency : Annotated[Optional[StrictStr], Field(description="The type of calculated output: Daily/Weekly/Monthly/Quarterly/Half-Yearly/Yearly.")] = None, metrics : Annotated[Optional[conlist(StrictStr)], Field(description="Determines what type of returns should be calculated, see https://support.lusid.com/knowledgebase/article/KA-01675/en-us for a list of available metrics.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, alternative_inc_date : Annotated[Optional[StrictStr], Field(description="The date from which to consider the Returns on the Portfolio, if this is different from the date when Returns begin. Can be a date string or Portfolio property.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfAggregatedReturn: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_aggregate_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], recipe_id_scope : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Scope for getting the fx rates")] = None, recipe_id_code : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Code for getting the fx rates")] = None, from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, composite_method : Annotated[Optional[StrictStr], Field(description="The method used to calculate the Portfolio performance: Equal/Asset.")] = None, period : Annotated[Optional[StrictStr], Field(description="The type of the returns used to calculate the aggregation result: Daily/Monthly.")] = None, output_frequency : Annotated[Optional[StrictStr], Field(description="The type of calculated output: Daily/Weekly/Monthly/Quarterly/Half-Yearly/Yearly.")] = None, metrics : Annotated[Optional[conlist(StrictStr)], Field(description="Determines what type of returns should be calculated, see https://support.lusid.com/knowledgebase/article/KA-01675/en-us for a list of available metrics.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, alternative_inc_date : Annotated[Optional[StrictStr], Field(description="The date from which to consider the Returns on the Portfolio, if this is different from the date when Returns begin. Can be a date string or Portfolio property.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfAggregatedReturn, Awaitable[ResourceListOfAggregatedReturn]]: # noqa: E501
"""[DEPRECATED] GetPortfolioAggregateReturns: Aggregate Returns (This is a deprecated endpoint). # noqa: E501
Aggregate Returns which are on the specified portfolio. # 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_portfolio_aggregate_returns(scope, code, return_scope, return_code, recipe_id_scope, recipe_id_code, from_effective_at, to_effective_at, composite_method, period, output_frequency, metrics, as_at, alternative_inc_date, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param recipe_id_scope: The Recipe Scope for getting the fx rates
:type recipe_id_scope: str
:param recipe_id_code: The Recipe Code for getting the fx rates
:type recipe_id_code: str
:param from_effective_at: The start date from which to calculate the Returns.
:type from_effective_at: str
:param to_effective_at: The end date for which to calculate the Returns.
:type to_effective_at: str
:param composite_method: The method used to calculate the Portfolio performance: Equal/Asset.
:type composite_method: str
:param period: The type of the returns used to calculate the aggregation result: Daily/Monthly.
:type period: str
:param output_frequency: The type of calculated output: Daily/Weekly/Monthly/Quarterly/Half-Yearly/Yearly.
:type output_frequency: str
:param metrics: Determines what type of returns should be calculated, see https://support.lusid.com/knowledgebase/article/KA-01675/en-us for a list of available metrics.
:type metrics: List[str]
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
:param alternative_inc_date: The date from which to consider the Returns on the Portfolio, if this is different from the date when Returns begin. Can be a date string or Portfolio property.
:type alternative_inc_date: 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: ResourceListOfAggregatedReturn
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_aggregate_returns_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_portfolio_aggregate_returns_with_http_info(scope, code, return_scope, return_code, recipe_id_scope, recipe_id_code, from_effective_at, to_effective_at, composite_method, period, output_frequency, metrics, as_at, alternative_inc_date, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_aggregate_returns_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], recipe_id_scope : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Scope for getting the fx rates")] = None, recipe_id_code : Annotated[Optional[constr(strict=True, max_length=64, min_length=1)], Field(description="The Recipe Code for getting the fx rates")] = None, from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, composite_method : Annotated[Optional[StrictStr], Field(description="The method used to calculate the Portfolio performance: Equal/Asset.")] = None, period : Annotated[Optional[StrictStr], Field(description="The type of the returns used to calculate the aggregation result: Daily/Monthly.")] = None, output_frequency : Annotated[Optional[StrictStr], Field(description="The type of calculated output: Daily/Weekly/Monthly/Quarterly/Half-Yearly/Yearly.")] = None, metrics : Annotated[Optional[conlist(StrictStr)], Field(description="Determines what type of returns should be calculated, see https://support.lusid.com/knowledgebase/article/KA-01675/en-us for a list of available metrics.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, alternative_inc_date : Annotated[Optional[StrictStr], Field(description="The date from which to consider the Returns on the Portfolio, if this is different from the date when Returns begin. Can be a date string or Portfolio property.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[DEPRECATED] GetPortfolioAggregateReturns: Aggregate Returns (This is a deprecated endpoint). # noqa: E501
Aggregate Returns which are on the specified portfolio. # 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_portfolio_aggregate_returns_with_http_info(scope, code, return_scope, return_code, recipe_id_scope, recipe_id_code, from_effective_at, to_effective_at, composite_method, period, output_frequency, metrics, as_at, alternative_inc_date, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param recipe_id_scope: The Recipe Scope for getting the fx rates
:type recipe_id_scope: str
:param recipe_id_code: The Recipe Code for getting the fx rates
:type recipe_id_code: str
:param from_effective_at: The start date from which to calculate the Returns.
:type from_effective_at: str
:param to_effective_at: The end date for which to calculate the Returns.
:type to_effective_at: str
:param composite_method: The method used to calculate the Portfolio performance: Equal/Asset.
:type composite_method: str
:param period: The type of the returns used to calculate the aggregation result: Daily/Monthly.
:type period: str
:param output_frequency: The type of calculated output: Daily/Weekly/Monthly/Quarterly/Half-Yearly/Yearly.
:type output_frequency: str
:param metrics: Determines what type of returns should be calculated, see https://support.lusid.com/knowledgebase/article/KA-01675/en-us for a list of available metrics.
:type metrics: List[str]
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
:param alternative_inc_date: The date from which to consider the Returns on the Portfolio, if this is different from the date when Returns begin. Can be a date string or Portfolio property.
:type alternative_inc_date: 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(ResourceListOfAggregatedReturn, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'return_scope',
'return_code',
'recipe_id_scope',
'recipe_id_code',
'from_effective_at',
'to_effective_at',
'composite_method',
'period',
'output_frequency',
'metrics',
'as_at',
'alternative_inc_date'
]
_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_portfolio_aggregate_returns" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
if _params['return_scope']:
_path_params['returnScope'] = _params['return_scope']
if _params['return_code']:
_path_params['returnCode'] = _params['return_code']
# process the query parameters
_query_params = []
if _params.get('recipe_id_scope') is not None: # noqa: E501
_query_params.append(('recipeIdScope', _params['recipe_id_scope']))
if _params.get('recipe_id_code') is not None: # noqa: E501
_query_params.append(('recipeIdCode', _params['recipe_id_code']))
if _params.get('from_effective_at') is not None: # noqa: E501
_query_params.append(('fromEffectiveAt', _params['from_effective_at']))
if _params.get('to_effective_at') is not None: # noqa: E501
_query_params.append(('toEffectiveAt', _params['to_effective_at']))
if _params.get('composite_method') is not None: # noqa: E501
_query_params.append(('compositeMethod', _params['composite_method']))
if _params.get('period') is not None: # noqa: E501
_query_params.append(('period', _params['period']))
if _params.get('output_frequency') is not None: # noqa: E501
_query_params.append(('outputFrequency', _params['output_frequency']))
if _params.get('metrics') is not None: # noqa: E501
_query_params.append(('metrics', _params['metrics']))
_collection_formats['metrics'] = 'multi'
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('alternative_inc_date') is not None: # noqa: E501
_query_params.append(('alternativeIncDate', _params['alternative_inc_date']))
# 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': "ResourceListOfAggregatedReturn",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/returns/{returnScope}/{returnCode}/aggregated', '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_portfolio_aggregated_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_request : Annotated[AggregatedReturnsRequest, Field(..., description="The request used in the AggregatedReturns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> AggregatedReturnsResponse: # noqa: E501
...
@overload
def get_portfolio_aggregated_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_request : Annotated[AggregatedReturnsRequest, Field(..., description="The request used in the AggregatedReturns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=True, **kwargs) -> AggregatedReturnsResponse: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_aggregated_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_request : Annotated[AggregatedReturnsRequest, Field(..., description="The request used in the AggregatedReturns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[AggregatedReturnsResponse, Awaitable[AggregatedReturnsResponse]]: # noqa: E501
"""[EARLY ACCESS] GetPortfolioAggregatedReturns: Aggregated Returns # noqa: E501
Aggregate Returns which are on the specified portfolio. # 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_portfolio_aggregated_returns(scope, code, aggregated_returns_request, from_effective_at, to_effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param aggregated_returns_request: The request used in the AggregatedReturns. (required)
:type aggregated_returns_request: AggregatedReturnsRequest
:param from_effective_at: The start date from which to calculate the Returns.
:type from_effective_at: str
:param to_effective_at: The end date for which to calculate the Returns.
:type to_effective_at: str
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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: AggregatedReturnsResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_aggregated_returns_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_portfolio_aggregated_returns_with_http_info(scope, code, aggregated_returns_request, from_effective_at, to_effective_at, as_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_aggregated_returns_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], aggregated_returns_request : Annotated[AggregatedReturnsRequest, Field(..., description="The request used in the AggregatedReturns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to calculate the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date for which to calculate the Returns.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetPortfolioAggregatedReturns: Aggregated Returns # noqa: E501
Aggregate Returns which are on the specified portfolio. # 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_portfolio_aggregated_returns_with_http_info(scope, code, aggregated_returns_request, from_effective_at, to_effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param aggregated_returns_request: The request used in the AggregatedReturns. (required)
:type aggregated_returns_request: AggregatedReturnsRequest
:param from_effective_at: The start date from which to calculate the Returns.
:type from_effective_at: str
:param to_effective_at: The end date for which to calculate the Returns.
:type to_effective_at: str
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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(AggregatedReturnsResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'aggregated_returns_request',
'from_effective_at',
'to_effective_at',
'as_at'
]
_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_portfolio_aggregated_returns" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('from_effective_at') is not None: # noqa: E501
_query_params.append(('fromEffectiveAt', _params['from_effective_at']))
if _params.get('to_effective_at') is not None: # noqa: E501
_query_params.append(('toEffectiveAt', _params['to_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']))
# 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['aggregated_returns_request'] is not None:
_body_params = _params['aggregated_returns_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': "AggregatedReturnsResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/returns/$aggregated', '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 get_portfolio_commands(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], from_as_at : Annotated[Optional[datetime], Field(description="The lower bound asAt datetime (inclusive) from which to retrieve commands. There is no lower bound if this is not specified.")] = None, to_as_at : Annotated[Optional[datetime], Field(description="The upper bound asAt datetime (inclusive) from which to retrieve commands. There is no upper bound if this is not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For example, to filter on the User ID, specify \"userId.id eq 'string'\". For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 500 if not specified.")] = None, **kwargs) -> ResourceListOfProcessedCommand: # noqa: E501
...
@overload
def get_portfolio_commands(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], from_as_at : Annotated[Optional[datetime], Field(description="The lower bound asAt datetime (inclusive) from which to retrieve commands. There is no lower bound if this is not specified.")] = None, to_as_at : Annotated[Optional[datetime], Field(description="The upper bound asAt datetime (inclusive) from which to retrieve commands. There is no upper bound if this is not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For example, to filter on the User ID, specify \"userId.id eq 'string'\". For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 500 if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfProcessedCommand: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_commands(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], from_as_at : Annotated[Optional[datetime], Field(description="The lower bound asAt datetime (inclusive) from which to retrieve commands. There is no lower bound if this is not specified.")] = None, to_as_at : Annotated[Optional[datetime], Field(description="The upper bound asAt datetime (inclusive) from which to retrieve commands. There is no upper bound if this is not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For example, to filter on the User ID, specify \"userId.id eq 'string'\". For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 500 if not specified.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfProcessedCommand, Awaitable[ResourceListOfProcessedCommand]]: # noqa: E501
"""GetPortfolioCommands: Get portfolio commands # noqa: E501
Get all the commands that modified a particular portfolio, including any input transactions. # 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_portfolio_commands(scope, code, from_as_at, to_as_at, filter, page, limit, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param from_as_at: The lower bound asAt datetime (inclusive) from which to retrieve commands. There is no lower bound if this is not specified.
:type from_as_at: datetime
:param to_as_at: The upper bound asAt datetime (inclusive) from which to retrieve commands. There is no upper bound if this is not specified.
:type to_as_at: datetime
:param filter: Expression to filter the results. For example, to filter on the User ID, specify \"userId.id eq 'string'\". For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.
:type filter: str
:param page: The pagination token to use to continue listing commands; this value is returned from the previous call.
:type page: str
:param limit: When paginating, limit the results to this number. Defaults to 500 if not specified.
:type limit: int
: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: ResourceListOfProcessedCommand
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_commands_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_portfolio_commands_with_http_info(scope, code, from_as_at, to_as_at, filter, page, limit, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_commands_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], from_as_at : Annotated[Optional[datetime], Field(description="The lower bound asAt datetime (inclusive) from which to retrieve commands. There is no lower bound if this is not specified.")] = None, to_as_at : Annotated[Optional[datetime], Field(description="The upper bound asAt datetime (inclusive) from which to retrieve commands. There is no upper bound if this is not specified.")] = None, filter : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression to filter the results. For example, to filter on the User ID, specify \"userId.id eq 'string'\". For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number. Defaults to 500 if not specified.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""GetPortfolioCommands: Get portfolio commands # noqa: E501
Get all the commands that modified a particular portfolio, including any input transactions. # 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_portfolio_commands_with_http_info(scope, code, from_as_at, to_as_at, filter, page, limit, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param from_as_at: The lower bound asAt datetime (inclusive) from which to retrieve commands. There is no lower bound if this is not specified.
:type from_as_at: datetime
:param to_as_at: The upper bound asAt datetime (inclusive) from which to retrieve commands. There is no upper bound if this is not specified.
:type to_as_at: datetime
:param filter: Expression to filter the results. For example, to filter on the User ID, specify \"userId.id eq 'string'\". For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.
:type filter: str
:param page: The pagination token to use to continue listing commands; this value is returned from the previous call.
:type page: str
:param limit: When paginating, limit the results to this number. Defaults to 500 if not specified.
:type limit: int
: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(ResourceListOfProcessedCommand, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'from_as_at',
'to_as_at',
'filter',
'page',
'limit'
]
_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_portfolio_commands" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('from_as_at') is not None: # noqa: E501
if isinstance(_params['from_as_at'], datetime):
_query_params.append(('fromAsAt', _params['from_as_at'].strftime(self.api_client.configuration.datetime_format)))
else:
_query_params.append(('fromAsAt', _params['from_as_at']))
if _params.get('to_as_at') is not None: # noqa: E501
if isinstance(_params['to_as_at'], datetime):
_query_params.append(('toAsAt', _params['to_as_at'].strftime(self.api_client.configuration.datetime_format)))
else:
_query_params.append(('toAsAt', _params['to_as_at']))
if _params.get('filter') is not None: # noqa: E501
_query_params.append(('filter', _params['filter']))
if _params.get('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
if _params.get('limit') is not None: # noqa: E501
_query_params.append(('limit', _params['limit']))
# 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': "ResourceListOfProcessedCommand",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/commands', '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_portfolio_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio Access Metadata Rule to retrieve.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which to retrieve the access metadata rule.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the portfolio access metadata.")] = None, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
def get_portfolio_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio Access Metadata Rule to retrieve.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], effective_at : Annotated[Optional[StrictStr], Field(description="The effectiveAt datetime at which to retrieve the access metadata rule.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the portfolio access metadata.")] = None, async_req: Optional[bool]=True, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
async def get_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, **kwargs) -> PortfolioProperties: # noqa: E501
...
@overload
def get_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> PortfolioProperties: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[PortfolioProperties, Awaitable[PortfolioProperties]]: # noqa: E501
"""GetPortfolioProperties: Get portfolio properties # noqa: E501
List all the properties of a particular portfolio. # 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_portfolio_properties(scope, code, effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.
:type as_at: datetime
: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: PortfolioProperties
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_properties_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_portfolio_properties_with_http_info(scope, code, effective_at, as_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_properties_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""GetPortfolioProperties: Get portfolio properties # noqa: E501
List all the properties of a particular portfolio. # 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_portfolio_properties_with_http_info(scope, code, effective_at, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.
:type as_at: datetime
: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(PortfolioProperties, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'effective_at',
'as_at'
]
_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_portfolio_properties" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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']))
# 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': "PortfolioProperties",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/properties', '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_portfolio_property_time_series(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_key : Annotated[StrictStr, Field(..., description="The property key of the property whose history to show. This must be from the 'Portfolio' domain and in the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], portfolio_effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime used to resolve the portfolio. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to show the history. Defaults to returning the current datetime if not supplied.")] = 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, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing properties; this value is returned from the previous call. If a pagination token is provided, the filter, portfolioEffectiveAt, and asAt fields must not have changed since the original request.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number.")] = None, **kwargs) -> ResourceListOfPropertyInterval: # noqa: E501
...
@overload
def get_portfolio_property_time_series(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_key : Annotated[StrictStr, Field(..., description="The property key of the property whose history to show. This must be from the 'Portfolio' domain and in the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], portfolio_effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime used to resolve the portfolio. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to show the history. Defaults to returning the current datetime if not supplied.")] = 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, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing properties; this value is returned from the previous call. If a pagination token is provided, the filter, portfolioEffectiveAt, and asAt fields must not have changed since the original request.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfPropertyInterval: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_property_time_series(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_key : Annotated[StrictStr, Field(..., description="The property key of the property whose history to show. This must be from the 'Portfolio' domain and in the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], portfolio_effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime used to resolve the portfolio. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to show the history. Defaults to returning the current datetime if not supplied.")] = 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, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing properties; this value is returned from the previous call. If a pagination token is provided, the filter, portfolioEffectiveAt, and asAt fields must not have changed since the original request.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfPropertyInterval, Awaitable[ResourceListOfPropertyInterval]]: # noqa: E501
"""[EARLY ACCESS] GetPortfolioPropertyTimeSeries: Get portfolio property time series # noqa: E501
Show the complete time series (history) for a particular portfolio property. # 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_portfolio_property_time_series(scope, code, property_key, portfolio_effective_at, as_at, filter, page, limit, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param property_key: The property key of the property whose history to show. This must be from the 'Portfolio' domain and in the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. (required)
:type property_key: str
:param portfolio_effective_at: The effective datetime used to resolve the portfolio. Defaults to the current LUSID system datetime if not specified.
:type portfolio_effective_at: str
:param as_at: The asAt datetime at which to show the history. Defaults to returning the current datetime if not supplied.
:type as_at: datetime
:param filter: Expression to filter the results. For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.
:type filter: str
:param page: The pagination token to use to continue listing properties; this value is returned from the previous call. If a pagination token is provided, the filter, portfolioEffectiveAt, and asAt fields must not have changed since the original request.
:type page: str
:param limit: When paginating, limit the results to this number.
:type limit: int
: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: ResourceListOfPropertyInterval
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_property_time_series_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_portfolio_property_time_series_with_http_info(scope, code, property_key, portfolio_effective_at, as_at, filter, page, limit, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_property_time_series_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], property_key : Annotated[StrictStr, Field(..., description="The property key of the property whose history to show. This must be from the 'Portfolio' domain and in the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], portfolio_effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime used to resolve the portfolio. Defaults to the current LUSID system datetime if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to show the history. Defaults to returning the current datetime if not supplied.")] = 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, see https://support.lusid.com/knowledgebase/article/KA-01914.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing properties; this value is returned from the previous call. If a pagination token is provided, the filter, portfolioEffectiveAt, and asAt fields must not have changed since the original request.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results to this number.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetPortfolioPropertyTimeSeries: Get portfolio property time series # noqa: E501
Show the complete time series (history) for a particular portfolio property. # 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_portfolio_property_time_series_with_http_info(scope, code, property_key, portfolio_effective_at, as_at, filter, page, limit, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param property_key: The property key of the property whose history to show. This must be from the 'Portfolio' domain and in the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. (required)
:type property_key: str
:param portfolio_effective_at: The effective datetime used to resolve the portfolio. Defaults to the current LUSID system datetime if not specified.
:type portfolio_effective_at: str
:param as_at: The asAt datetime at which to show the history. Defaults to returning the current datetime if not supplied.
:type as_at: datetime
:param filter: Expression to filter the results. For more information about filtering, see https://support.lusid.com/knowledgebase/article/KA-01914.
:type filter: str
:param page: The pagination token to use to continue listing properties; this value is returned from the previous call. If a pagination token is provided, the filter, portfolioEffectiveAt, and asAt fields must not have changed since the original request.
:type page: str
:param limit: When paginating, limit the results to this number.
:type limit: int
: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(ResourceListOfPropertyInterval, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'property_key',
'portfolio_effective_at',
'as_at',
'filter',
'page',
'limit'
]
_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_portfolio_property_time_series" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('property_key') is not None: # noqa: E501
_query_params.append(('propertyKey', _params['property_key']))
if _params.get('portfolio_effective_at') is not None: # noqa: E501
_query_params.append(('portfolioEffectiveAt', _params['portfolio_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('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
if _params.get('limit') is not None: # noqa: E501
_query_params.append(('limit', _params['limit']))
# 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': "ResourceListOfPropertyInterval",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/properties/time-series', '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_portfolio_relations(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve relations. 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 relations. Defaults to returning 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 the relations. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These must be from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. Only identifier types provided will be used to look up relevant entities in relations. If not applicable, provide an empty array.")] = None, **kwargs) -> ResourceListOfRelation: # noqa: E501
...
@overload
def get_portfolio_relations(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve relations. 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 relations. Defaults to returning 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 the relations. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These must be from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. Only identifier types provided will be used to look up relevant entities in relations. If not applicable, provide an empty array.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfRelation: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_relations(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve relations. 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 relations. Defaults to returning 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 the relations. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These must be from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. Only identifier types provided will be used to look up relevant entities in relations. If not applicable, provide an empty array.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfRelation, Awaitable[ResourceListOfRelation]]: # noqa: E501
"""[EXPERIMENTAL] GetPortfolioRelations: Get portfolio relations # noqa: E501
Get relations for a particular portfolio. # 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_portfolio_relations(scope, code, effective_at, as_at, filter, identifier_types, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to retrieve relations. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to retrieve relations. Defaults to returning the latest LUSID AsAt time if not specified.
:type as_at: datetime
:param filter: Expression to filter the relations. Provide a null or empty string for this field until further notice.
:type filter: str
:param identifier_types: Identifier types (as property keys) used for referencing Persons or Legal Entities. These must be from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. Only identifier types provided will be used to look up relevant entities in relations. 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: ResourceListOfRelation
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_relations_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_portfolio_relations_with_http_info(scope, code, effective_at, as_at, filter, identifier_types, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_relations_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to retrieve relations. 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 relations. Defaults to returning 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 the relations. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These must be from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. Only identifier types provided will be used to look up relevant entities in relations. If not applicable, provide an empty array.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EXPERIMENTAL] GetPortfolioRelations: Get portfolio relations # noqa: E501
Get relations for a particular portfolio. # 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_portfolio_relations_with_http_info(scope, code, effective_at, as_at, filter, identifier_types, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to retrieve relations. Defaults to the current LUSID system datetime if not specified.
:type effective_at: str
:param as_at: The asAt datetime at which to retrieve relations. Defaults to returning the latest LUSID AsAt time if not specified.
:type as_at: datetime
:param filter: Expression to filter the relations. Provide a null or empty string for this field until further notice.
:type filter: str
:param identifier_types: Identifier types (as property keys) used for referencing Persons or Legal Entities. These must be from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. Only identifier types provided will be used to look up relevant entities in relations. 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(ResourceListOfRelation, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'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_portfolio_relations" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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('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': "ResourceListOfRelation",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/relations', '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_portfolio_relationships(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to retrieve 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 returning 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 the relationships. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These can be specified from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. An Empty array may be used to return all related Entities.")] = None, **kwargs) -> ResourceListOfRelationship: # noqa: E501
...
@overload
def get_portfolio_relationships(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to retrieve 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 returning 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 the relationships. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These can be specified from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. An Empty array may be used to return all related Entities.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfRelationship: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_relationships(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to retrieve 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 returning 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 the relationships. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These can be specified from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. An Empty array may be used to return all related Entities.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfRelationship, Awaitable[ResourceListOfRelationship]]: # noqa: E501
"""[EARLY ACCESS] GetPortfolioRelationships: Get portfolio relationships # noqa: E501
Get relationships for a particular portfolio. # 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_portfolio_relationships(scope, code, effective_at, as_at, filter, identifier_types, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to retrieve 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 returning the latest LUSID AsAt time if not specified.
:type as_at: datetime
:param filter: Expression to filter the relationships. Provide a null or empty string for this field until further notice.
:type filter: str
:param identifier_types: Identifier types (as property keys) used for referencing Persons or Legal Entities. These can be specified from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. An Empty array may be used to return all related Entities.
: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_portfolio_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_portfolio_relationships_with_http_info(scope, code, effective_at, as_at, filter, identifier_types, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_relationships_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to retrieve 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 returning 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 the relationships. Provide a null or empty string for this field until further notice.")] = None, identifier_types : Annotated[Optional[conlist(StrictStr)], Field(description="Identifier types (as property keys) used for referencing Persons or Legal Entities. These can be specified from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. An Empty array may be used to return all related Entities.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetPortfolioRelationships: Get portfolio relationships # noqa: E501
Get relationships for a particular portfolio. # 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_portfolio_relationships_with_http_info(scope, code, effective_at, as_at, filter, identifier_types, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to retrieve 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 returning the latest LUSID AsAt time if not specified.
:type as_at: datetime
:param filter: Expression to filter the relationships. Provide a null or empty string for this field until further notice.
:type filter: str
:param identifier_types: Identifier types (as property keys) used for referencing Persons or Legal Entities. These can be specified from the 'Person' or 'LegalEntity' domains and have the format {domain}/{scope}/{code}, for example 'Person/CompanyDetails/Role'. An Empty array may be used to return all related Entities.
: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 = [
'scope',
'code',
'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_portfolio_relationships" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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('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/portfolios/{scope}/{code}/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 get_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to get the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date from which to get the Returns.")] = None, period : Annotated[Optional[StrictStr], Field(description="Show the Returns on a Daily or Monthly period. Defaults to Daily.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> ResourceListOfPerformanceReturn: # noqa: E501
...
@overload
def get_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to get the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date from which to get the Returns.")] = None, period : Annotated[Optional[StrictStr], Field(description="Show the Returns on a Daily or Monthly period. Defaults to Daily.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfPerformanceReturn: # noqa: E501
...
[docs]
@validate_arguments
def get_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to get the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date from which to get the Returns.")] = None, period : Annotated[Optional[StrictStr], Field(description="Show the Returns on a Daily or Monthly period. Defaults to Daily.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfPerformanceReturn, Awaitable[ResourceListOfPerformanceReturn]]: # noqa: E501
"""[EARLY ACCESS] GetPortfolioReturns: Get Returns # noqa: E501
Get Returns which are on the specified portfolio. # 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_portfolio_returns(scope, code, return_scope, return_code, from_effective_at, to_effective_at, period, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param from_effective_at: The start date from which to get the Returns.
:type from_effective_at: str
:param to_effective_at: The end date from which to get the Returns.
:type to_effective_at: str
:param period: Show the Returns on a Daily or Monthly period. Defaults to Daily.
:type period: str
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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: ResourceListOfPerformanceReturn
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the get_portfolio_returns_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_portfolio_returns_with_http_info(scope, code, return_scope, return_code, from_effective_at, to_effective_at, period, as_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def get_portfolio_returns_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], from_effective_at : Annotated[Optional[StrictStr], Field(description="The start date from which to get the Returns.")] = None, to_effective_at : Annotated[Optional[StrictStr], Field(description="The end date from which to get the Returns.")] = None, period : Annotated[Optional[StrictStr], Field(description="Show the Returns on a Daily or Monthly period. Defaults to Daily.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the Returns. Defaults to the latest.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] GetPortfolioReturns: Get Returns # noqa: E501
Get Returns which are on the specified portfolio. # 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_portfolio_returns_with_http_info(scope, code, return_scope, return_code, from_effective_at, to_effective_at, period, as_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param from_effective_at: The start date from which to get the Returns.
:type from_effective_at: str
:param to_effective_at: The end date from which to get the Returns.
:type to_effective_at: str
:param period: Show the Returns on a Daily or Monthly period. Defaults to Daily.
:type period: str
:param as_at: The asAt datetime at which to retrieve the Returns. Defaults to the latest.
:type as_at: datetime
: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(ResourceListOfPerformanceReturn, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'return_scope',
'return_code',
'from_effective_at',
'to_effective_at',
'period',
'as_at'
]
_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_portfolio_returns" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
if _params['return_scope']:
_path_params['returnScope'] = _params['return_scope']
if _params['return_code']:
_path_params['returnCode'] = _params['return_code']
# process the query parameters
_query_params = []
if _params.get('from_effective_at') is not None: # noqa: E501
_query_params.append(('fromEffectiveAt', _params['from_effective_at']))
if _params.get('to_effective_at') is not None: # noqa: E501
_query_params.append(('toEffectiveAt', _params['to_effective_at']))
if _params.get('period') is not None: # noqa: E501
_query_params.append(('period', _params['period']))
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']))
# 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': "ResourceListOfPerformanceReturn",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/returns/{returnScope}/{returnCode}', '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_portfolios_access_metadata_by_key(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio Access Metadata Rule to retrieve.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio")], metadata_key : Annotated[constr(strict=True, max_length=256, min_length=1), Field(..., description="Key of the metadata to retrieve")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective date of the rule")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the portfolio access metadata.")] = None, **kwargs) -> List[AccessMetadataValue]: # noqa: E501
...
@overload
def get_portfolios_access_metadata_by_key(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio Access Metadata Rule to retrieve.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio")], metadata_key : Annotated[constr(strict=True, max_length=256, min_length=1), Field(..., description="Key of the metadata to retrieve")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective date of the rule")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the portfolio access metadata.")] = None, async_req: Optional[bool]=True, **kwargs) -> List[AccessMetadataValue]: # noqa: E501
...
@overload
async def list_instrument_event_instructions(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instructions. Defaults to latest if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing instructions; 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, 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 to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, **kwargs) -> PagedResourceListOfInstrumentEventInstruction: # noqa: E501
...
@overload
def list_instrument_event_instructions(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instructions. Defaults to latest if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing instructions; 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, 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 to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, async_req: Optional[bool]=True, **kwargs) -> PagedResourceListOfInstrumentEventInstruction: # noqa: E501
...
[docs]
@validate_arguments
def list_instrument_event_instructions(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instructions. Defaults to latest if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing instructions; 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, 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 to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[PagedResourceListOfInstrumentEventInstruction, Awaitable[PagedResourceListOfInstrumentEventInstruction]]: # noqa: E501
"""[EARLY ACCESS] ListInstrumentEventInstructions: List Instrument Event Instructions # noqa: E501
Lists all instructions for a particular portfolio # 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_instrument_event_instructions(scope, code, portfolio_effective_at, as_at, page, limit, filter, sort_by, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: str
:param as_at: The asAt datetime at which to retrieve the instructions. Defaults to latest if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing instructions; 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 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 to sort by, each suffixed by \" ASC\" or \" DESC\".
:type sort_by: 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: PagedResourceListOfInstrumentEventInstruction
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the list_instrument_event_instructions_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_instrument_event_instructions_with_http_info(scope, code, portfolio_effective_at, as_at, page, limit, filter, sort_by, **kwargs) # noqa: E501
[docs]
@validate_arguments
def list_instrument_event_instructions_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, as_at : Annotated[Optional[datetime], Field(description="The asAt datetime at which to retrieve the instructions. Defaults to latest if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing instructions; 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, 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 to sort by, each suffixed by \" ASC\" or \" DESC\".")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] ListInstrumentEventInstructions: List Instrument Event Instructions # noqa: E501
Lists all instructions for a particular portfolio # 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_instrument_event_instructions_with_http_info(scope, code, portfolio_effective_at, as_at, page, limit, filter, sort_by, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: str
:param as_at: The asAt datetime at which to retrieve the instructions. Defaults to latest if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing instructions; 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 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 to sort by, each suffixed by \" ASC\" or \" DESC\".
:type sort_by: 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(PagedResourceListOfInstrumentEventInstruction, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'portfolio_effective_at',
'as_at',
'page',
'limit',
'filter',
'sort_by'
]
_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_instrument_event_instructions" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('portfolio_effective_at') is not None: # noqa: E501
_query_params.append(('portfolioEffectiveAt', _params['portfolio_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('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
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'
# 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': "PagedResourceListOfInstrumentEventInstruction",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/instrumenteventinstructions', '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_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results per page to this number.")] = None, **kwargs) -> ResourceListOfProperty: # noqa: E501
...
@overload
def list_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results per page to this number.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfProperty: # noqa: E501
...
[docs]
@validate_arguments
def list_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results per page to this number.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfProperty, Awaitable[ResourceListOfProperty]]: # noqa: E501
"""[EARLY ACCESS] ListPortfolioProperties: Get portfolio properties # noqa: E501
List all the properties of a particular portfolio. # 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_portfolio_properties(scope, code, effective_at, as_at, page, limit, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing commands; this value is returned from the previous call.
:type page: str
:param limit: When paginating, limit the results per page to this number.
:type limit: int
: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: ResourceListOfProperty
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the list_portfolio_properties_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_portfolio_properties_with_http_info(scope, code, effective_at, as_at, page, limit, **kwargs) # noqa: E501
[docs]
@validate_arguments
def list_portfolio_properties_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], effective_at : Annotated[Optional[constr(strict=True, max_length=256, min_length=0)], Field(description="The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing commands; this value is returned from the previous call.")] = None, limit : Annotated[Optional[conint(strict=True, le=5000, ge=1)], Field(description="When paginating, limit the results per page to this number.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] ListPortfolioProperties: Get portfolio properties # noqa: E501
List all the properties of a particular portfolio. # 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_portfolio_properties_with_http_info(scope, code, effective_at, as_at, page, limit, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param effective_at: The effective datetime or cut label at which to list the portfolio's properties. 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 portfolio's properties. Defaults to returning the latest version of each property if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing commands; this value is returned from the previous call.
:type page: str
:param limit: When paginating, limit the results per page to this number.
:type limit: int
: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(ResourceListOfProperty, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'effective_at',
'as_at',
'page',
'limit'
]
_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_portfolio_properties" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
if _params.get('limit') is not None: # noqa: E501
_query_params.append(('limit', _params['limit']))
# 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': "ResourceListOfProperty",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/properties/list', '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_portfolios(self, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios; 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, 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 example, to filter on the transaction type, specify \"type eq 'Transaction'\". 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, query : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression specifying the criteria that the returned portfolios must meet. For example, to see which portfolios have holdings in instruments with a LusidInstrumentId (LUID) of 'LUID_PPA8HI6M' or a Figi of 'BBG000BLNNH6', specify \"instrument.identifiers in (('LusidInstrumentId', 'LUID_PPA8HI6M'), ('Figi', 'BBG000BLNNH6'))\".")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ResourceListOfPortfolio: # noqa: E501
...
@overload
def list_portfolios(self, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios; 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, 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 example, to filter on the transaction type, specify \"type eq 'Transaction'\". 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, query : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression specifying the criteria that the returned portfolios must meet. For example, to see which portfolios have holdings in instruments with a LusidInstrumentId (LUID) of 'LUID_PPA8HI6M' or a Figi of 'BBG000BLNNH6', specify \"instrument.identifiers in (('LusidInstrumentId', 'LUID_PPA8HI6M'), ('Figi', 'BBG000BLNNH6'))\".")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfPortfolio: # noqa: E501
...
[docs]
@validate_arguments
def list_portfolios(self, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios; 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, 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 example, to filter on the transaction type, specify \"type eq 'Transaction'\". 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, query : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression specifying the criteria that the returned portfolios must meet. For example, to see which portfolios have holdings in instruments with a LusidInstrumentId (LUID) of 'LUID_PPA8HI6M' or a Figi of 'BBG000BLNNH6', specify \"instrument.identifiers in (('LusidInstrumentId', 'LUID_PPA8HI6M'), ('Figi', 'BBG000BLNNH6'))\".")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfPortfolio, Awaitable[ResourceListOfPortfolio]]: # noqa: E501
"""ListPortfolios: List portfolios # noqa: E501
List all the portfolios 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_portfolios(effective_at, as_at, page, limit, filter, sort_by, query, property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param effective_at: The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing portfolios; 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 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 example, to filter on the transaction type, specify \"type eq 'Transaction'\". 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 query: Expression specifying the criteria that the returned portfolios must meet. For example, to see which portfolios have holdings in instruments with a LusidInstrumentId (LUID) of 'LUID_PPA8HI6M' or a Figi of 'BBG000BLNNH6', specify \"instrument.identifiers in (('LusidInstrumentId', 'LUID_PPA8HI6M'), ('Figi', 'BBG000BLNNH6'))\".
:type query: str
:param property_keys: A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the portfolios 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: ResourceListOfPortfolio
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the list_portfolios_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_portfolios_with_http_info(effective_at, as_at, page, limit, filter, sort_by, query, property_keys, relationship_definition_ids, **kwargs) # noqa: E501
[docs]
@validate_arguments
def list_portfolios_with_http_info(self, effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios; 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, 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 example, to filter on the transaction type, specify \"type eq 'Transaction'\". 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, query : Annotated[Optional[constr(strict=True, max_length=16384, min_length=0)], Field(description="Expression specifying the criteria that the returned portfolios must meet. For example, to see which portfolios have holdings in instruments with a LusidInstrumentId (LUID) of 'LUID_PPA8HI6M' or a Figi of 'BBG000BLNNH6', specify \"instrument.identifiers in (('LusidInstrumentId', 'LUID_PPA8HI6M'), ('Figi', 'BBG000BLNNH6'))\".")] = None, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""ListPortfolios: List portfolios # noqa: E501
List all the portfolios 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_portfolios_with_http_info(effective_at, as_at, page, limit, filter, sort_by, query, property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param effective_at: The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing portfolios; 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 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 example, to filter on the transaction type, specify \"type eq 'Transaction'\". 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 query: Expression specifying the criteria that the returned portfolios must meet. For example, to see which portfolios have holdings in instruments with a LusidInstrumentId (LUID) of 'LUID_PPA8HI6M' or a Figi of 'BBG000BLNNH6', specify \"instrument.identifiers in (('LusidInstrumentId', 'LUID_PPA8HI6M'), ('Figi', 'BBG000BLNNH6'))\".
:type query: str
:param property_keys: A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the portfolios 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(ResourceListOfPortfolio, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'effective_at',
'as_at',
'page',
'limit',
'filter',
'sort_by',
'query',
'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_portfolios" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
# 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('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
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('query') is not None: # noqa: E501
_query_params.append(('query', _params['query']))
if _params.get('property_keys') is not None: # noqa: E501
_query_params.append(('propertyKeys', _params['property_keys']))
_collection_formats['propertyKeys'] = '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': "ResourceListOfPortfolio",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios', '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_portfolios_for_scope(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope whose portfolios to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios. 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, 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 example, to return only transactions with a transaction type of 'Buy', specify \"type eq 'Buy'\". 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, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ResourceListOfPortfolio: # noqa: E501
...
@overload
def list_portfolios_for_scope(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope whose portfolios to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios. 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, 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 example, to return only transactions with a transaction type of 'Buy', specify \"type eq 'Buy'\". 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, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfPortfolio: # noqa: E501
...
[docs]
@validate_arguments
def list_portfolios_for_scope(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope whose portfolios to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios. 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, 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 example, to return only transactions with a transaction type of 'Buy', specify \"type eq 'Buy'\". 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, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[ResourceListOfPortfolio, Awaitable[ResourceListOfPortfolio]]: # noqa: E501
"""ListPortfoliosForScope: List portfolios for scope # noqa: E501
List all the portfolios in a particular scope. # 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_portfolios_for_scope(scope, effective_at, as_at, page, limit, filter, sort_by, property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param scope: The scope whose portfolios to list. (required)
:type scope: str
:param effective_at: The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing portfolios. 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 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 example, to return only transactions with a transaction type of 'Buy', specify \"type eq 'Buy'\". 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 property_keys: A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the portfolios 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: ResourceListOfPortfolio
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the list_portfolios_for_scope_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_portfolios_for_scope_with_http_info(scope, effective_at, as_at, page, limit, filter, sort_by, property_keys, relationship_definition_ids, **kwargs) # noqa: E501
[docs]
@validate_arguments
def list_portfolios_for_scope_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope whose portfolios to list.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.")] = None, page : Annotated[Optional[constr(strict=True, max_length=500, min_length=1)], Field(description="The pagination token to use to continue listing portfolios. 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, 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 example, to return only transactions with a transaction type of 'Buy', specify \"type eq 'Buy'\". 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, property_keys : Annotated[Optional[conlist(StrictStr)], Field(description="A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 portfolios in the response. These must take the form {relationshipDefinitionScope}/{relationshipDefinitionCode}.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""ListPortfoliosForScope: List portfolios for scope # noqa: E501
List all the portfolios in a particular scope. # 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_portfolios_for_scope_with_http_info(scope, effective_at, as_at, page, limit, filter, sort_by, property_keys, relationship_definition_ids, async_req=True)
>>> result = thread.get()
:param scope: The scope whose portfolios to list. (required)
:type scope: str
:param effective_at: The effective datetime or cut label at which to list the portfolios. 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 portfolios. Defaults to returning the latest version of each portfolio if not specified.
:type as_at: datetime
:param page: The pagination token to use to continue listing portfolios. 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 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 example, to return only transactions with a transaction type of 'Buy', specify \"type eq 'Buy'\". 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 property_keys: A list of property keys from the 'Portfolio' domain to decorate onto each portfolio, or 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 property_keys: List[str]
:param relationship_definition_ids: A list of relationship definitions that are used to decorate related entities onto the portfolios 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(ResourceListOfPortfolio, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'effective_at',
'as_at',
'page',
'limit',
'filter',
'sort_by',
'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_portfolios_for_scope" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
# 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('page') is not None: # noqa: E501
_query_params.append(('page', _params['page']))
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('property_keys') is not None: # noqa: E501
_query_params.append(('propertyKeys', _params['property_keys']))
_collection_formats['propertyKeys'] = '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': "ResourceListOfPortfolio",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}', '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_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], operation : Annotated[conlist(Operation), Field(..., description="The json patch document. For more check: https://datatracker.ietf.org/doc/html/rfc6902.")], **kwargs) -> Portfolio: # noqa: E501
...
@overload
def patch_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], operation : Annotated[conlist(Operation), Field(..., description="The json patch document. For more check: https://datatracker.ietf.org/doc/html/rfc6902.")], async_req: Optional[bool]=True, **kwargs) -> Portfolio: # noqa: E501
...
[docs]
@validate_arguments
def patch_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], operation : Annotated[conlist(Operation), Field(..., description="The json patch document. For more check: https://datatracker.ietf.org/doc/html/rfc6902.")], async_req: Optional[bool]=None, **kwargs) -> Union[Portfolio, Awaitable[Portfolio]]: # noqa: E501
"""[EARLY ACCESS] PatchPortfolio: Patch portfolio. # noqa: E501
Create or update certain fields for a particular portfolio. The behaviour is defined by the JSON Patch specification. Currently supported are: CreationDate, InstrumentScopes. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.patch_portfolio(scope, code, operation, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param operation: The json patch document. For more check: https://datatracker.ietf.org/doc/html/rfc6902. (required)
:type operation: List[Operation]
: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: Portfolio
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the patch_portfolio_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.patch_portfolio_with_http_info(scope, code, operation, **kwargs) # noqa: E501
[docs]
@validate_arguments
def patch_portfolio_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], operation : Annotated[conlist(Operation), Field(..., description="The json patch document. For more check: https://datatracker.ietf.org/doc/html/rfc6902.")], **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] PatchPortfolio: Patch portfolio. # noqa: E501
Create or update certain fields for a particular portfolio. The behaviour is defined by the JSON Patch specification. Currently supported are: CreationDate, InstrumentScopes. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.patch_portfolio_with_http_info(scope, code, operation, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param operation: The json patch document. For more check: https://datatracker.ietf.org/doc/html/rfc6902. (required)
:type operation: List[Operation]
: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(Portfolio, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'operation'
]
_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 patch_portfolio" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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['operation'] is not None:
_body_params = _params['operation']
# 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': "Portfolio",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}', 'PATCH',
_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_portfolio_access_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio Access Metadata Rule.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], access_metadata_operation : Annotated[conlist(AccessMetadataOperation), Field(..., description="The Json Patch document")], effective_at : Annotated[Optional[StrictStr], Field(description="The date this rule will effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective date until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' date of the Access Metadata")] = None, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
def patch_portfolio_access_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio Access Metadata Rule.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], access_metadata_operation : Annotated[conlist(AccessMetadataOperation), Field(..., description="The Json Patch document")], effective_at : Annotated[Optional[StrictStr], Field(description="The date this rule will effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective date until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' date of the Access Metadata")] = None, async_req: Optional[bool]=True, **kwargs) -> Dict[str, List[AccessMetadataValue]]: # noqa: E501
...
@overload
async def update_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], update_portfolio_request : Annotated[UpdatePortfolioRequest, Field(..., description="The updated portfolio definition.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to update the definition. Defaults to the current LUSID system datetime if not specified.")] = None, **kwargs) -> Portfolio: # noqa: E501
...
@overload
def update_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], update_portfolio_request : Annotated[UpdatePortfolioRequest, Field(..., description="The updated portfolio definition.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to update the definition. Defaults to the current LUSID system datetime if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> Portfolio: # noqa: E501
...
[docs]
@validate_arguments
def update_portfolio(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], update_portfolio_request : Annotated[UpdatePortfolioRequest, Field(..., description="The updated portfolio definition.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to update the definition. Defaults to the current LUSID system datetime if not specified.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[Portfolio, Awaitable[Portfolio]]: # noqa: E501
"""UpdatePortfolio: Update portfolio # noqa: E501
Update the definition of a particular portfolio. Note that not all elements of a portfolio definition are modifiable due to the potential implications for data already stored. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_portfolio(scope, code, update_portfolio_request, effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param update_portfolio_request: The updated portfolio definition. (required)
:type update_portfolio_request: UpdatePortfolioRequest
:param effective_at: The effective datetime or cut label at which to update the definition. Defaults to the current LUSID system datetime if not specified.
:type effective_at: 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: Portfolio
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the update_portfolio_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.update_portfolio_with_http_info(scope, code, update_portfolio_request, effective_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def update_portfolio_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], update_portfolio_request : Annotated[UpdatePortfolioRequest, Field(..., description="The updated portfolio definition.")], effective_at : Annotated[Optional[StrictStr], Field(description="The effective datetime or cut label at which to update the definition. Defaults to the current LUSID system datetime if not specified.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""UpdatePortfolio: Update portfolio # noqa: E501
Update the definition of a particular portfolio. Note that not all elements of a portfolio definition are modifiable due to the potential implications for data already stored. # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.update_portfolio_with_http_info(scope, code, update_portfolio_request, effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param update_portfolio_request: The updated portfolio definition. (required)
:type update_portfolio_request: UpdatePortfolioRequest
:param effective_at: The effective datetime or cut label at which to update the definition. Defaults to the current LUSID system datetime if not specified.
:type effective_at: 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(Portfolio, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'update_portfolio_request',
'effective_at'
]
_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 update_portfolio" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('effective_at') is not None: # noqa: E501
_query_params.append(('effectiveAt', _params['effective_at']))
# 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['update_portfolio_request'] is not None:
_body_params = _params['update_portfolio_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': "Portfolio",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}', 'PUT',
_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_instrument_event_instructions(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], 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 (default)")], request_body : Annotated[Dict[str, InstrumentEventInstructionRequest], Field(..., description="The instructions to be upserted to the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, **kwargs) -> InstrumentEventInstructionsResponse: # noqa: E501
...
@overload
def upsert_instrument_event_instructions(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], 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 (default)")], request_body : Annotated[Dict[str, InstrumentEventInstructionRequest], Field(..., description="The instructions to be upserted to the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, async_req: Optional[bool]=True, **kwargs) -> InstrumentEventInstructionsResponse: # noqa: E501
...
[docs]
@validate_arguments
def upsert_instrument_event_instructions(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], 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 (default)")], request_body : Annotated[Dict[str, InstrumentEventInstructionRequest], Field(..., description="The instructions to be upserted to the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[InstrumentEventInstructionsResponse, Awaitable[InstrumentEventInstructionsResponse]]: # noqa: E501
"""[EARLY ACCESS] UpsertInstrumentEventInstructions: Upsert Instrument Event Instructions # noqa: E501
Batch upsert for instrument event instructions, for the portfolio or individual holdings # 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_instrument_event_instructions(scope, code, success_mode, request_body, portfolio_effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param success_mode: Whether the batch request should fail atomically or in a partial fashion - allowed values: Atomic, Partial (default) (required)
:type success_mode: str
:param request_body: The instructions to be upserted to the portfolio. (required)
:type request_body: Dict[str, InstrumentEventInstructionRequest]
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: 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: InstrumentEventInstructionsResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the upsert_instrument_event_instructions_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_instrument_event_instructions_with_http_info(scope, code, success_mode, request_body, portfolio_effective_at, **kwargs) # noqa: E501
[docs]
@validate_arguments
def upsert_instrument_event_instructions_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], 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 (default)")], request_body : Annotated[Dict[str, InstrumentEventInstructionRequest], Field(..., description="The instructions to be upserted to the portfolio.")], portfolio_effective_at : Annotated[Optional[constr(strict=True, max_length=6000, min_length=0)], Field(description="The effective date at which the portfolio will be resolved. Defaults to current time if not specified.")] = None, **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] UpsertInstrumentEventInstructions: Upsert Instrument Event Instructions # noqa: E501
Batch upsert for instrument event instructions, for the portfolio or individual holdings # 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_instrument_event_instructions_with_http_info(scope, code, success_mode, request_body, portfolio_effective_at, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param success_mode: Whether the batch request should fail atomically or in a partial fashion - allowed values: Atomic, Partial (default) (required)
:type success_mode: str
:param request_body: The instructions to be upserted to the portfolio. (required)
:type request_body: Dict[str, InstrumentEventInstructionRequest]
:param portfolio_effective_at: The effective date at which the portfolio will be resolved. Defaults to current time if not specified.
:type portfolio_effective_at: 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(InstrumentEventInstructionsResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'success_mode',
'request_body',
'portfolio_effective_at'
]
_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_instrument_event_instructions" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# process the query parameters
_query_params = []
if _params.get('portfolio_effective_at') is not None: # noqa: E501
_query_params.append(('portfolioEffectiveAt', _params['portfolio_effective_at']))
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': "InstrumentEventInstructionsResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/instrumenteventinstructions', '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_portfolio_access_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope to use when updating or inserting the Portfolio Access Metadata Rule.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], metadata_key : Annotated[constr(strict=True, max_length=256, min_length=1), Field(..., description="Key of the access metadata to upsert")], upsert_portfolio_access_metadata_request : Annotated[UpsertPortfolioAccessMetadataRequest, Field(..., description="The Portfolio Access Metadata Rule to update or insert")], effective_at : Annotated[Optional[StrictStr], Field(description="The date this rule will effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective date until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' date of the Access Metadata")] = None, **kwargs) -> ResourceListOfAccessMetadataValueOf: # noqa: E501
...
@overload
def upsert_portfolio_access_metadata(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope to use when updating or inserting the Portfolio Access Metadata Rule.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="Portfolio code")], metadata_key : Annotated[constr(strict=True, max_length=256, min_length=1), Field(..., description="Key of the access metadata to upsert")], upsert_portfolio_access_metadata_request : Annotated[UpsertPortfolioAccessMetadataRequest, Field(..., description="The Portfolio Access Metadata Rule to update or insert")], effective_at : Annotated[Optional[StrictStr], Field(description="The date this rule will effective from")] = None, effective_until : Annotated[Optional[datetime], Field(description="The effective date until which the Access Metadata is valid. If not supplied this will be valid indefinitely, or until the next 'effectiveAt' date of the Access Metadata")] = None, async_req: Optional[bool]=True, **kwargs) -> ResourceListOfAccessMetadataValueOf: # noqa: E501
...
@overload
async def upsert_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], request_body : Annotated[Dict[str, ModelProperty], Field(..., description="The properties to be created or updated. Each property in the request must be keyed by its unique property key. This has the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], **kwargs) -> PortfolioProperties: # noqa: E501
...
@overload
def upsert_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], request_body : Annotated[Dict[str, ModelProperty], Field(..., description="The properties to be created or updated. Each property in the request must be keyed by its unique property key. This has the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], async_req: Optional[bool]=True, **kwargs) -> PortfolioProperties: # noqa: E501
...
[docs]
@validate_arguments
def upsert_portfolio_properties(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], request_body : Annotated[Dict[str, ModelProperty], Field(..., description="The properties to be created or updated. Each property in the request must be keyed by its unique property key. This has the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], async_req: Optional[bool]=None, **kwargs) -> Union[PortfolioProperties, Awaitable[PortfolioProperties]]: # noqa: E501
"""UpsertPortfolioProperties: Upsert portfolio properties # noqa: E501
Create or update one or more properties for a particular portfolio. A property is updated if it already exists and created if it does not. All properties must be from the 'Portfolio' domain. Properties have an <i>effectiveFrom</i> datetime from which the property is valid, and an <i>effectiveUntil</i> datetime until which it is valid. Not supplying an <i>effectiveUntil</i> datetime results in the property being valid indefinitely, or until the next <i>effectiveFrom</i> datetime of the property. # 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_portfolio_properties(scope, code, request_body, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param request_body: The properties to be created or updated. Each property in the request must be keyed by its unique property key. This has the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. (required)
:type request_body: Dict[str, ModelProperty]
: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: PortfolioProperties
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the upsert_portfolio_properties_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_portfolio_properties_with_http_info(scope, code, request_body, **kwargs) # noqa: E501
[docs]
@validate_arguments
def upsert_portfolio_properties_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the portfolio. Together with the scope this uniquely identifies the portfolio.")], request_body : Annotated[Dict[str, ModelProperty], Field(..., description="The properties to be created or updated. Each property in the request must be keyed by its unique property key. This has the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'.")], **kwargs) -> ApiResponse: # noqa: E501
"""UpsertPortfolioProperties: Upsert portfolio properties # noqa: E501
Create or update one or more properties for a particular portfolio. A property is updated if it already exists and created if it does not. All properties must be from the 'Portfolio' domain. Properties have an <i>effectiveFrom</i> datetime from which the property is valid, and an <i>effectiveUntil</i> datetime until which it is valid. Not supplying an <i>effectiveUntil</i> datetime results in the property being valid indefinitely, or until the next <i>effectiveFrom</i> datetime of the property. # 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_portfolio_properties_with_http_info(scope, code, request_body, async_req=True)
>>> result = thread.get()
:param scope: The scope of the portfolio. (required)
:type scope: str
:param code: The code of the portfolio. Together with the scope this uniquely identifies the portfolio. (required)
:type code: str
:param request_body: The properties to be created or updated. Each property in the request must be keyed by its unique property key. This has the format {domain}/{scope}/{code}, for example 'Portfolio/Manager/Id'. (required)
:type request_body: Dict[str, ModelProperty]
: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(PortfolioProperties, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'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_portfolio_properties" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
# 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['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': "PortfolioProperties",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/properties', '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_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], performance_return : Annotated[conlist(PerformanceReturn), Field(..., description="This contains the Returns which need to be upsert.")], **kwargs) -> UpsertReturnsResponse: # noqa: E501
...
@overload
def upsert_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], performance_return : Annotated[conlist(PerformanceReturn), Field(..., description="This contains the Returns which need to be upsert.")], async_req: Optional[bool]=True, **kwargs) -> UpsertReturnsResponse: # noqa: E501
...
[docs]
@validate_arguments
def upsert_portfolio_returns(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], performance_return : Annotated[conlist(PerformanceReturn), Field(..., description="This contains the Returns which need to be upsert.")], async_req: Optional[bool]=None, **kwargs) -> Union[UpsertReturnsResponse, Awaitable[UpsertReturnsResponse]]: # noqa: E501
"""[EARLY ACCESS] UpsertPortfolioReturns: Upsert Returns # noqa: E501
Update or insert returns into the specified portfolio. # 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_portfolio_returns(scope, code, return_scope, return_code, performance_return, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param performance_return: This contains the Returns which need to be upsert. (required)
:type performance_return: List[PerformanceReturn]
: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: UpsertReturnsResponse
"""
kwargs['_return_http_data_only'] = True
if '_preload_content' in kwargs:
message = "Error! Please call the upsert_portfolio_returns_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_portfolio_returns_with_http_info(scope, code, return_scope, return_code, performance_return, **kwargs) # noqa: E501
[docs]
@validate_arguments
def upsert_portfolio_returns_with_http_info(self, scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Portfolio.")], code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Portfolio.")], return_scope : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The scope of the Returns.")], return_code : Annotated[constr(strict=True, max_length=64, min_length=1), Field(..., description="The code of the Returns.")], performance_return : Annotated[conlist(PerformanceReturn), Field(..., description="This contains the Returns which need to be upsert.")], **kwargs) -> ApiResponse: # noqa: E501
"""[EARLY ACCESS] UpsertPortfolioReturns: Upsert Returns # noqa: E501
Update or insert returns into the specified portfolio. # 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_portfolio_returns_with_http_info(scope, code, return_scope, return_code, performance_return, async_req=True)
>>> result = thread.get()
:param scope: The scope of the Portfolio. (required)
:type scope: str
:param code: The code of the Portfolio. (required)
:type code: str
:param return_scope: The scope of the Returns. (required)
:type return_scope: str
:param return_code: The code of the Returns. (required)
:type return_code: str
:param performance_return: This contains the Returns which need to be upsert. (required)
:type performance_return: List[PerformanceReturn]
: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(UpsertReturnsResponse, status_code(int), headers(HTTPHeaderDict))
"""
_params = locals()
_all_params = [
'scope',
'code',
'return_scope',
'return_code',
'performance_return'
]
_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_portfolio_returns" % _key
)
_params[_key] = _val
del _params['kwargs']
_collection_formats = {}
# process the path parameters
_path_params = {}
if _params['scope']:
_path_params['scope'] = _params['scope']
if _params['code']:
_path_params['code'] = _params['code']
if _params['return_scope']:
_path_params['returnScope'] = _params['return_scope']
if _params['return_code']:
_path_params['returnCode'] = _params['return_code']
# 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['performance_return'] is not None:
_body_params = _params['performance_return']
# 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': "UpsertReturnsResponse",
'400': "LusidValidationProblemDetails",
}
return self.api_client.call_api(
'/api/portfolios/{scope}/{code}/returns/{returnScope}/{returnCode}', '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'))