"""API Client for ows-abacus-contract.""" import requests class AbacusContractAPIClient: """API Client for ows-abacus-contract.""" def __init__(self, base_url, headers): """Initialize ows_contract_APIClient class.""" self.base_url = base_url self.headers = headers def post_contract(self, body): """Execute a POST against /contract endpoint.""" endpoint = '{}/contract'.format(self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contracts_by_search_term(self, params): """Execute a GET against /contracts endpoint.""" endpoint = '{}/contracts'.format(self.base_url) return requests.get(endpoint, headers=self.headers, params=params) def get_contract(self, contract_id): """Execute a GET against /contract/ endpoint.""" endpoint = '{}/contract/{}'.format(self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def put_contract(self, contract_id, body): """Execute a PUT against /contract/ endpoint.""" endpoint = '{}/contract/{}'.format(self.base_url, contract_id) return requests.put(endpoint, headers=self.headers, json=body) def get_contracts(self, params=None): """Execute a GET against /contracts endpoint.""" endpoint = '{}/contracts'.format(self.base_url) return requests.get(endpoint, headers=self.headers, params=params) def post_contracts(self, body=None): """Execute a POST against /contracts endpoint (acts like GET).""" endpoint = '{}/contracts'.format(self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def generate_contract_body( self, contract_name, account_id, contract_type='distribution', reference_signing_entity_id=2 ): """Generate a body for POST /contract endpoint.""" return {'contract_name': contract_name, 'contract_type': contract_type, 'account_id': account_id, 'reference_signing_entity_id': reference_signing_entity_id, 'term_start': '2019-06-01', 'term_end': '2020-06-01', 'summary_note': 'testing', 'general_note': 'testing' } def post_contract_term(self, contract_id, body): """Execute a POST against /contract//contract-term.""" endpoint = '{}/contract/{}/contract-term'.format(self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_terms(self, contract_id): """Execute a GET against /contracts//contract_terms.""" endpoint = '{}/contracts/{}/contract-terms'.format(self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def generate_contract_term_conditions(self, priority=1, term_rate='80.00', comission='20.00', countries=['USA'], stores=['some-store-1', 'some-another'], transaction_types=['type-1', 'type-1-1']): """Generate a body for POST contract-term//conditions.""" return { 'priority': priority, 'term_rate': term_rate, 'commission': comission, 'conditions': { 'countries': countries, 'stores': stores, 'transaction_types': transaction_types} } def post_contract_term_conditions(self, contract_term_id, body): """Execute a POST against /contract-term//conditions.""" endpoint = '{}/contract-term/{}/conditions'.format( self.base_url, contract_term_id) return requests.post(endpoint, headers=self.headers, json=body) def put_contract_term_conditions(self, contract_term_id, body): """Execute a PUT against /contract-term//conditions.""" endpoint = '{}/contract-term/{}/conditions'.format( self.base_url, contract_term_id) return requests.put(endpoint, headers=self.headers, json=body) def get_contract_term_conditions(self, contract_term_id): """Execute a GET against /contract-term//conditions.""" endpoint = '{}/contract-term/{}/conditions'.format( self.base_url, contract_term_id) return requests.get(endpoint, headers=self.headers) def soft_delete_contract_term_condition(self, body): """Execute a PUT against /contract-term-conditions/soft-delete.""" endpoint = '{}/contract-term-conditions/soft-delete'.format( self.base_url) return requests.put(endpoint, headers=self.headers, json=body) def put_contract_term(self, contract_term_id, body): """Execute a PUT against /contract-term/.""" endpoint = '{}/contract-term/{}'.format( self.base_url, contract_term_id) return requests.put(endpoint, headers=self.headers, json=body) def get_reference_transaction_types(self): """Execute a GET against /reference-transaction-types.""" endpoint = '{}/reference-transaction-types'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_contract_term(self, contract_term_id): """Execute a GET against /contract-term/.""" endpoint = '{}/contract-term/{}'.format( self.base_url, contract_term_id) return requests.get(endpoint, headers=self.headers) def get_tsv_contract_snapshot(self, body): """Execute a POST against /contracts/snapshot endpoint.""" endpoint = '{}/contracts/snapshot'.format(self.base_url) return requests.post(endpoint, json=body) def get_tsv_contract_terms_snapshot(self, body): """Execute a POST against /contract-terms/snapshot endpoint.""" endpoint = '{}/contract-terms/snapshot'.format(self.base_url) return requests.post(endpoint, json=body) def get_contracts_by_oa_contract_id(self, body): """Execute a POST against /oa-contracts/contracts.""" endpoint = '{}/oa-contracts/contracts'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def post_contract_from_template(self, template_id, body): """Execute a POST against /contract-template/{template_id}/contract/.""" endpoint = '{}/contract-template/{}/contract/'.format( self.base_url, template_id) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_terms_by_account_id(self, account_id, body): """GET contract terms by account_id.""" endpoint = '{}/account/{}/contract-terms/'.format( self.base_url, account_id) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_vat_info_by_contract_id(self, body): """POST /contracts/vat-info .""" endpoint = '{}/contracts/vat-info'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_reserves_by_contract_ids(self, body): """GET reserves by list of contract_ids.""" endpoint = '{}/contract/contract-reserves/'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_reserves_batch(self, body): """GET reserves by list of contract_reserve_ids.""" endpoint = '{}/contract-reserves/'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def post_contract_reserves(self, contract_id, body): """POST reserve by contract id.""" endpoint = '{}/contract/{}/reserves/'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_reserves(self, contract_id): """GET reserve by contract id.""" endpoint = '{}/contract/{}/reserves/'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def put_contract_reserves(self, contract_id, body): """PUT reserve by contract id.""" endpoint = '{}/contract/{}/reserves/'.format( self.base_url, contract_id) return requests.put(endpoint, headers=self.headers, json=body) def post_contract_advance(self, contract_id, body): """POST advance by contract id.""" endpoint = '{}/contract/{}/advance/'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_advances(self, contract_id, advance_status): """GET advance by contract id.""" endpoint = '{}/contract/{}/advances/{}'.format( self.base_url, contract_id, advance_status) return requests.get(endpoint, headers=self.headers) def get_contract_advances_by_id(self, contract_advance_id): """GET advance by contract_advance_id.""" endpoint = '{}/contract-advance/{}/'.format( self.base_url, contract_advance_id) return requests.get(endpoint, headers=self.headers) def put_contract_advance(self, contract_advance_id, body): """PUT contract_advance.""" endpoint = '{}/contract-advance/{}/'.format( self.base_url, contract_advance_id) return requests.put(endpoint, headers=self.headers, json=body) def delete_contract_advances(self, contract_advance_id): """DELETE advance by contract_advance_id.""" endpoint = '{}/contract-advance/{}/'.format( self.base_url, contract_advance_id) return requests.delete(endpoint, headers=self.headers) def post_contract_exclusions(self, contract_id, body): """POST /contract/{contract_id}/exclusions .""" endpoint = '{}/contract/{}/exclusions'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_exclusions(self, contract_id): """GET /contract/{contract_id}/exclusions .""" endpoint = '{}/contract/{}/exclusions'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def put_contract_exclusions(self, contract_id, body): """PUT /contract/{contract_id}/exclusions .""" endpoint = '{}/contract/{}/exclusions'.format( self.base_url, contract_id) return requests.put(endpoint, headers=self.headers, json=body) def get_contracts_by_list_of_contract_ids(self, body): """POST /contracts .""" endpoint = '{}/contracts'.format(self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contracts_by_account_id(self, account_id): """GET /contracts/account/{account_id} .""" endpoint = '{}/contracts/account/{}'.format(self.base_url, account_id) return requests.get(endpoint, headers=self.headers) def get_contracts_by_list_of_account_ids(self, body): """POST /contracts/accounts/ .""" endpoint = '{}/contracts/accounts'.format(self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contracts_by_account_id_dataloaded(self, body): """POST /contracts/account/dataloader .""" endpoint = '{}/contracts/account/dataloader'.format(self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def post_legacy_contract(self, contract_id, body): """POST /contract/{contract_id}/legacy-contract .""" endpoint = '{}/contract/{}/legacy-contract'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def get_legacy_contract(self, contract_id): """GET /contract/{contract_id}/legacy-contract .""" endpoint = '{}/contract/{}/legacy-contract'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def post_contract_party(self, body): """POST /contract-party .""" endpoint = '{}/contract-party'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_parties_by_type(self, contract_id, target_type): """GET /contract/{contract_id}/{target_type}/parties/ .""" endpoint = '{}/contract/{}/{}/parties/'.format( self.base_url, contract_id, target_type) return requests.get(endpoint, headers=self.headers) def get_contract_parties_by_contract_id(self, contract_id): """GET /contract/{contract_id}/parties/ .""" endpoint = '{}/contract/{}/parties/'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def delete_contract_party(self, contract_party_id): """DELETE /contract-party/ .""" endpoint = '{}/contract-party/{}'.format( self.base_url, contract_party_id) return requests.delete(endpoint, headers=self.headers) def get_contract_term_schedules(self, contract_term_id): """GET /contract-term//contract-term-schedules .""" endpoint = '{}/contract-term/{}/contract-term-schedules'.format( self.base_url, contract_term_id) return requests.get(endpoint, headers=self.headers) def delete_contract_term(self, contract_term_id): """DELETE advance by contract_advance_id.""" endpoint = '{}/contract-term/{}/'.format( self.base_url, contract_term_id) return requests.delete(endpoint, headers=self.headers) def get_reference_payment_types(self): """GET /reference-payment-types .""" endpoint = '{}/reference-payment-types'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_reference_payment_type_by_id(self, reference_payment_type_id): """GET /reference-payment-type/ .""" endpoint = '{}/reference-payment-type/{}'.format( self.base_url, reference_payment_type_id) return requests.get(endpoint, headers=self.headers) def get_reference_signing_entities(self): """GET /reference-signing-entities .""" endpoint = '{}/reference-signing-entities'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_reference_signing_entity_by_id(self, reference_signing_entity_id): """GET /reference-signing-entity/ .""" endpoint = '{}/reference-signing-entity/{}'.format( self.base_url, reference_signing_entity_id) return requests.get(endpoint, headers=self.headers) def get_reference_payment_entities(self): """GET /reference-payment-entities .""" endpoint = '{}/reference-payment-entities'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_reference_payment_entity_by_id(self, reference_payment_entity_id): """GET /reference-payment-entity/ .""" endpoint = '{}/reference-payment-entity/{}'.format( self.base_url, reference_payment_entity_id) return requests.get(endpoint, headers=self.headers) def get_transaction_type_groups_by_group_admin(self, group_admin): """GET /reference-transaction-type-group/group-admin/ .""" endpoint = '{}/reference-transaction-type-group/group-admin/{}'.format( self.base_url, group_admin) return requests.get(endpoint, headers=self.headers) def get_transaction_types_by_parent_group( self, reference_transaction_type_group_id ): """GET /reference-transaction-type-group//reference-transaction-types/ .""" endpoint = ('{}/reference-transaction-type-group/' '{}/reference-transaction-types/').format( self.base_url, reference_transaction_type_group_id) return requests.get(endpoint, headers=self.headers) def post_contract_lifecycle_schedule_detail_dataloader(self, body): """POST /contract-lifecycle-schedule-detail/dataloader .""" endpoint = '{}/contract-lifecycle-schedule-detail/dataloader'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_lifecycle_schedule_detail_by_id( self, contract_lifecycle_schedule_detail_id): """GET /contract-lifecycle-schedule-detail/ .""" # noqa endpoint = '{}/contract-lifecycle-schedule-detail/{}'.format( self.base_url, contract_lifecycle_schedule_detail_id) return requests.get(endpoint, headers=self.headers) def get_contract_lifecycle_schedules_by_contract_id( self, contract_id): """GET /contract//contract-lifecycle-schedules .""" endpoint = '{}/contract/{}/contract-lifecycle-schedules'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def get_contract_lifecycle_schedule_by_id( self, contract_lifecycle_schedule_id): """GET /contract-lifecycle-schedule// .""" endpoint = '{}/contract-lifecycle-schedule/{}'.format( self.base_url, contract_lifecycle_schedule_id) return requests.get(endpoint, headers=self.headers) def post_contract_lifecycle_schedule_dataloader(self, body): """POST /contract-lifecycle-schedules/dataloader .""" endpoint = '{}/contract-lifecycle-schedules/dataloader'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_lifecycle_by_contract_id(self, contract_id): """GET /contracts/{contract_id}/contract-lifecycle .""" endpoint = '{}/contract/{}/contract-lifecycle/'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def get_contract_lifecycle_by_contract_lifecycle_schedule_id( self, contract_lifecycle_schedule_id ): """GET /contract-lifecycle-schedule/{contract_lifecycle_schedule_id}/contract-lifecycle .""" # noqa endpoint = '{}/contract-lifecycle-schedule/{}/contract-lifecycle/'.format( self.base_url, contract_lifecycle_schedule_id) return requests.get(endpoint, headers=self.headers) def get_contract_lifecycle_by_contract_lifecycle_id(self, contract_lifecycle_id): """GET /contract-lifecycle/{contract_lifecycle_id}/ .""" endpoint = '{}/contract-lifecycle/{}/'.format( self.base_url, contract_lifecycle_id) return requests.get(endpoint, headers=self.headers) def post_contract_lifecycle_dataloader(self, body): """POST /contract-lifecycles/dataloader .""" endpoint = '{}/contract-lifecycles/dataloader'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def post_contract_lifecycle_schedules(self, contract_id, body): """POST /contract/{contract_id}/contract-lifecycle-schedules/ .""" endpoint = '{}/contract/{}/contract-lifecycle-schedules/'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def post_contract_lifecycle( self, contract_id, contract_lifecycle_schedule_id, body ): """POST /contract/{contract_id}/contract-lifecycle-schedule/{contract_lifecycle_schedule_id}/contract-lifecycle .""" # noqa endpoint = ( f'{self.base_url}' f'/contract/{contract_id}' f'/contract-lifecycle-schedule/{contract_lifecycle_schedule_id}' f'/contract-lifecycle' ) return requests.post(endpoint, headers=self.headers, json=body) def put_contract_lifecycle_schedules(self, contract_id, body): """PUT /contract/{contract_id}/contract-lifecycle-schedules/ .""" endpoint = '{}/contract/{}/contract-lifecycle-schedules/'.format( self.base_url, contract_id) return requests.put(endpoint, headers=self.headers, json=body) def post_contract_with_lifecycle_schedules_and_contract_lifecycle( self, body ): """POST /contract/contract-lifecycle-schedule/contract-lifecycle .""" # noqa endpoint = '{}/contract/contract-lifecycle-schedule/contract-lifecycle'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def put_contract_terminate(self, contract_id, body): """PUT /contract//terminate/ .""" endpoint = '{}/contract/{}/terminate/'.format( self.base_url, contract_id) return requests.put(endpoint, headers=self.headers, json=body) def put_reactivate_contract( self, contract_id ): """PUT /contract//reactivate/ .""" # noqa endpoint = '{}/contract/{}/reactivate/'.format( self.base_url, contract_id) return requests.put(endpoint, headers=self.headers) def get_reference_sap_profit_centers(self): """GET /reference-sap-profit-centers .""" endpoint = '{}/reference-sap-profit-center/'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_reference_sap_profit_center_by_id( self, reference_sap_profit_center_id ): """GET /reference-sap-profit-center/ .""" endpoint = '{}/reference-sap-profit-center/{}'.format( self.base_url, reference_sap_profit_center_id) return requests.get(endpoint, headers=self.headers) def get_renewable_contract_lifecycles(self): """GET /contract-lifecycles/renewable/ .""" endpoint = '{}/contract-lifecycles/renewable/'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_activable_contract_lifecycles(self): """GET /contract-lifecycles/activable/ .""" endpoint = '{}/contract-lifecycles/activable/'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_terrminable_contract_lifecycles(self): """GET /contract-lifecycles/terminable/ .""" endpoint = '{}/contract-lifecycles/terminable/'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def post_contract_lifecycles_renew(self, contract_ids): """POST /contract-lifecycles/renew/ .""" endpoint = '{}/contract-lifecycles/renew/'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=[contract_ids]) def post_contract_lifecycles_activate(self, contract_ids): """POST /contract-lifecycles/activate .""" endpoint = '{}/contract-lifecycles/activate'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=[contract_ids]) def post_contract_lifecycles_terminate(self, contract_ids): """POST /contract-lifecycles/terminate .""" endpoint = '{}/contract-lifecycles/terminate'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=[contract_ids]) def get_mechanical_rates(self): """GET /reference-mechanical-rates/ .""" endpoint = '{}/reference-mechanical-rates/'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_mechanical_rate_by_id(self, reference_mechanical_rate_id): """GET /reference-mechanical-rate// .""" endpoint = '{}/reference-mechanical-rate/{}'.format( self.base_url, reference_mechanical_rate_id) return requests.get(endpoint, headers=self.headers) def get_mechanical_deduction_by_id(self, contract_mechanical_deduction_id): """GET /reference-mechanical-deduction// .""" endpoint = '{}/contract-mechanical-deduction/{}/'.format( self.base_url, contract_mechanical_deduction_id) return requests.get(endpoint, headers=self.headers) def get_mechanical_deductions_by_contract_id(self, contract_id): """GET /contract//contract-mechanical-deductions/.""" endpoint = '{}/contract/{}/contract-mechanical-deductions/'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def post_mechanical_deductions_worldwide(self, contract_id, body): """POST /contract//contract-mechanical-deductions/worldwide/.""" endpoint = '{}/contract/{}/contract-mechanical-deductions/worldwide/'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def post_mechanical_deduction(self, contract_id, body): """POST /contract//contract-mechanical-deduction/.""" endpoint = '{}/contract/{}/contract-mechanical-deduction/'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def put_mechanical_deduction(self, contract_mechanical_deduction_id, body): """PUT /contract-mechanical-deduction/ .""" endpoint = '{}/contract-mechanical-deduction/{}'.format( self.base_url, contract_mechanical_deduction_id) return requests.put(endpoint, headers=self.headers, json=body) def post_contract_contract_lifecycle_dataloader(self, body): """POST /contract/contract-lifecycle/dataloader .""" endpoint = '{}/contract/contract-lifecycle/dataloader'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_transaction_types_groups_and_transaction_types(self): """GET /reference-transaction-type-group/reference_transaction_type_group_transaction_types.""" # noqa endpoint = '{}/reference-transaction-type-group/reference_transaction_type_group_transaction_types'.format( # noqa self.base_url) return requests.get(endpoint, headers=self.headers) def get_transaction_type_groups(self): """GET /reference-transaction-type-groups .""" endpoint = f'{self.base_url}/reference-transaction-type-groups' return requests.get(endpoint, headers=self.headers) def delete_mechanical_deduction(self, deduction_id): """DELETE /contract-mechanical-deduction/{deduction_id}.""" # noqa endpoint = '{}/contract-mechanical-deduction/{}'.format( # noqa self.base_url, deduction_id) return requests.delete(endpoint, headers=self.headers) def post_contracts_dataloader(self, body): """POST /contracts/dataloader/ .""" endpoint = '{}/contracts/dataloader/'.format( self.base_url) return requests.post(endpoint, headers=self.headers, json=body) def get_contract_flowthrough_by_flowthrough_id(self, flowthrough_id): """GET /contract-flowthrough/ .""" endpoint = '{}/contract-flowthrough/{}'.format( self.base_url, flowthrough_id) return requests.get(endpoint, headers=self.headers) def put_contract_flowthrough(self, flowthrough_id, body): """PUT /contract-flowthrough/ .""" endpoint = '{}/contract-flowthrough/{}'.format( self.base_url, flowthrough_id) return requests.put(endpoint, headers=self.headers, json=body) def delete_contract_flowthrough(self, flowthrough_id): """DELETE /contract-flowthrough/ .""" endpoint = '{}/contract-flowthrough/{}'.format( self.base_url, flowthrough_id) return requests.delete(endpoint, headers=self.headers) def get_contract_flowthrough_by_contract_id(self, contract_id): """GET /contract//contract-flowthrough/ .""" endpoint = '{}/contract/{}/contract-flowthrough/'.format( self.base_url, contract_id) return requests.get(endpoint, headers=self.headers) def post_contract_flowthrough(self, contract_id, body): """POST /contract//contract-flowthrough/ .""" endpoint = '{}/contract/{}/contract-flowthrough/'.format( self.base_url, contract_id) return requests.post(endpoint, headers=self.headers, json=body) def get_reference_flowthrough_calculations(self): """GET /reference-flowthrough-calculations/ .""" endpoint = '{}/reference-flowthrough-calculations/'.format( self.base_url) return requests.get(endpoint, headers=self.headers) def get_reference_flowthrough_calculations_by_id( self, reference_flowthrough_calculation_id ): """GET /reference-flowthrough-calculations/ .""" # noqa endpoint = '{}/reference-flowthrough-calculation/{}'.format( self.base_url, reference_flowthrough_calculation_id) return requests.get(endpoint, headers=self.headers)