SAP BAPI with pyrfc

sap_bapi.py

import pandas as pd
from pyrfc import Connection, ABAPApplicationError, ABAPRuntimeError, LogonError, CommunicationError


class RFC_Handle(object):
    def __init__(self):
        self.conn = None

    def login(self, sap_id, sap_pwd):
        try:
            self.sap_id = sap_id
            self.conn = Connection(user=self.sap_id, passwd=sap_pwd, ashost='xxx.xxx.xxx.xxx', sysnr='xx', client='xxx')

            return self
        except CommunicationError:
            print("Could not connect to server.")
            raise
        except LogonError:
            print("Could not log in. Please Check your ID/PW.")
            raise
        except Exception as err:
            print(str(err))

    def cost_estimation_list(self, material, valid_date, costing_status):
        costing_variant = [{'SIGN': 'I', 'OPTION': 'EQ', 'LOW': 'XXXX'}]
        material = [{'SIGN': 'I', 'OPTION': 'EQ', 'LOW': material}]
        plant = [{'SIGN': 'I', 'OPTION': 'EQ', 'LOW': 'XXXX'}]
        validity_from = [{'SIGN': 'I', 'OPTION': 'BT', 'LOW': valid_date[0], 'HIGH': valid_date[1]}]
        costing_status = [{'SIGN': 'I', 'OPTION': 'EQ', 'LOW': x} for x in costing_status.split(', ')]

        cost_estimations = self.conn.call('BAPI_COSTESTIMATE_GETLIST',
                                          COST_ESTIMATE='01',
                                          MAXROWS=0,
                                          COSTING_VARIANT=costing_variant,
                                          MATERIAL=material,
                                          PLANT=plant,
                                          VALIDITY_FROM=validity_from,
                                          COSTING_STATUS=costing_status,
                                          COSTING_VERSION=[],
                                          COST_ESTIMATE_LIST=[],
                                          COSTINGRUN=[],
                                          COSTINGRUN_DATE=[],
                                          )

        return pd.DataFrame(cost_estimations['COST_ESTIMATE_LIST']).sort_values(by='VALID_TO', ascending=False)

    # Check BAPI_COSTESTIMATE_GETEXPLOTION also.
    def cost_itemization(self, cost_estimate):
        itemized = self.conn.call('BAPI_COSTESTIMATE_ITEMIZATION',
                                  REFERENCEOBJECT=cost_estimate['REF_OBJECT'],
                                  COSTINGNUMBER=cost_estimate['CSTG_NUM'],
                                  COSTINGTYPE=cost_estimate['CSTG_TYPE'],
                                  COSTINGDATE=cost_estimate['CSTG_DATE'],
                                  COSTINGVERSION=cost_estimate['VERSION'],
                                  VALUATIONVARIANT=cost_estimate['VLTN_VRNT'],
                                  ENTEREDMANUALLY=cost_estimate['ENTER_MAN'],
                                  )

        return pd.DataFrame(itemized['ITEMIZATION_LIST'])