Coverage for mindsdb / integrations / handlers / sap_erp_handler / sap_erp_handler.py: 0%
32 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1from collections import OrderedDict
2from mindsdb_sql_parser import parse_sql
4from mindsdb.integrations.libs.const import HANDLER_CONNECTION_ARG_TYPE as ARG_TYPE
5from mindsdb.integrations.handlers.sap_erp_handler.api import SAPERP
6from mindsdb.integrations.handlers.sap_erp_handler.sap_erp_tables import (
7 AddressEmailAddressTable,
8 AddressFaxNumberTable,
9 AddressHomePageURLTable,
10 AddressPhoneNumberTable,
11 BPAddrDepdntIntlLocNumberTable,
12 BPContactToAddressTable,
13 BPContactToFuncAndDeptTable,
14 BPCreditWorthinessTable,
15 BPDataControllerTable,
16 BPFinancialServicesExtnTable,
17 BPFinancialServicesReportingTable,
18 BPFiscalYearInformationTable,
19 BPRelationshipTable,
20 BuPaAddressUsageTable,
21 BuPaIdentificationTable,
22 BuPaIndustryTable,
23 BusinessPartnerTable,
24 BusinessPartnerAddressTable,
25 BusinessPartnerContactTable,
26 BusinessPartnerPaymentCardTable,
27 BusinessPartnerRatingTable,
28 BusinessPartnerRoleTable,
29 BusinessPartnerTaxNumberTable,
30 BusPartAddrDepdntTaxNumberTable,
31 CustAddrDepdntExtIdentifierTable,
32 CustAddrDepdntInformationTable,
33 CustomerCompanyTable,
34 CustomerCompanyTextTable,
35 CustomerDunningTable,
36 CustomerSalesAreaTable,
37 CustomerSalesAreaTaxTable,
38 CustomerSalesAreaTextTable,
39 CustomerTaxGroupingTable,
40 CustomerTextTable,
41 CustomerUnloadingPointTable,
42 CustomerWithHoldingTaxTable,
43 CustSalesPartnerFuncTable,
44 CustSlsAreaAddrDepdntInfoTable,
45 CustSlsAreaAddrDepdntTaxInfoTable,
46 CustUnldgPtAddrDepdntInfoTable,
47 SupplierTable,
48 SupplierCompanyTable,
49 SupplierCompanyTextTable,
50 SupplierDunningTable,
51 SupplierPartnerFuncTable,
52 SupplierPurchasingOrgTable,
53 SupplierPurchasingOrgTextTable,
54 SupplierTextTable,
55 SupplierWithHoldingTaxTable,
56)
58from mindsdb.integrations.libs.api_handler import APIHandler
59from mindsdb.integrations.libs.response import HandlerStatusResponse as StatusResponse
62class SAPERPHandler(APIHandler):
64 def __init__(self, name: str, **kwargs) -> None:
65 super().__init__(name)
66 self.connection = None
67 self.is_connected = False
68 self.api_key = kwargs.get("connection_data", {}).get("api_key", "")
69 self.base_url = kwargs.get("connection_data", {}).get("base_url", "")
70 _tables = [
71 AddressEmailAddressTable,
72 AddressFaxNumberTable,
73 AddressHomePageURLTable,
74 AddressPhoneNumberTable,
75 BPAddrDepdntIntlLocNumberTable,
76 BPContactToAddressTable,
77 BPContactToFuncAndDeptTable,
78 BPCreditWorthinessTable,
79 BPDataControllerTable,
80 BPFinancialServicesExtnTable,
81 BPFinancialServicesReportingTable,
82 BPFiscalYearInformationTable,
83 BPRelationshipTable,
84 BuPaAddressUsageTable,
85 BuPaIdentificationTable,
86 BuPaIndustryTable,
87 BusinessPartnerTable,
88 BusinessPartnerAddressTable,
89 BusinessPartnerContactTable,
90 BusinessPartnerPaymentCardTable,
91 BusinessPartnerRatingTable,
92 BusinessPartnerRoleTable,
93 BusinessPartnerTaxNumberTable,
94 BusPartAddrDepdntTaxNumberTable,
95 CustAddrDepdntExtIdentifierTable,
96 CustAddrDepdntInformationTable,
97 CustomerCompanyTable,
98 CustomerCompanyTextTable,
99 CustomerDunningTable,
100 CustomerSalesAreaTable,
101 CustomerSalesAreaTaxTable,
102 CustomerSalesAreaTextTable,
103 CustomerTaxGroupingTable,
104 CustomerTextTable,
105 CustomerUnloadingPointTable,
106 CustomerWithHoldingTaxTable,
107 CustSalesPartnerFuncTable,
108 CustSlsAreaAddrDepdntInfoTable,
109 CustSlsAreaAddrDepdntTaxInfoTable,
110 CustUnldgPtAddrDepdntInfoTable,
111 SupplierTable,
112 SupplierCompanyTable,
113 SupplierCompanyTextTable,
114 SupplierDunningTable,
115 SupplierPartnerFuncTable,
116 SupplierPurchasingOrgTable,
117 SupplierPurchasingOrgTextTable,
118 SupplierTextTable,
119 SupplierWithHoldingTaxTable,
120 ]
121 for Table in _tables:
122 self._register_table(Table.name, Table(self))
123 self.connect()
125 def check_connection(self) -> StatusResponse:
126 resp = StatusResponse(False)
127 if self.connection and not self.connection.is_connected():
128 resp.error = "Client not connected"
129 else:
130 resp.success = True
131 return resp
133 def connect(self) -> SAPERP:
134 self.connection = SAPERP(self.base_url, self.api_key)
135 return self.connection
137 def native_query(self, query: str) -> StatusResponse:
138 ast = parse_sql(query)
139 return self.query(ast)
142connection_args = OrderedDict(
143 api_key={
144 'type': ARG_TYPE.STR,
145 'description': 'API Token for accessing SAP ERP',
146 'required': True,
147 'label': 'API Key',
148 },
149 base_url={
150 'type': ARG_TYPE.STR,
151 'description': 'Base URL of the host',
152 'required': True,
153 'label': 'Base URL',
154 }
155)
158connection_args_example = OrderedDict(
159 api_key='23d6b9e0c2fab7eba2e8b7e452cead47',
160 base_url='https://sandbox.api.sap.com/s4hanacloud/sap/opu/odata/sap/API_BUSINESS_PARTNER/'
161)