SAP table access and report generation with pyrfc

sap_table_access_and_report_generation.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 table_access(self, table_name, options, fields):
        # for qs in [options, fields]:
        #     self.query_syntax_check(qs)

        try:
            if table_name not in self.tables_used:
                self.tables_used.append(table_name)

            delimiter = '^'

            if options not in ('', '*'):
                # Solution for TEXT option's 72 characters limitation.
                options = [{'TEXT': x} for x in textwrap.wrap(options, 72)]
            else:
                options = ''

            if fields not in ('', '*'):
                fields = [{'FIELDNAME': x} for x in fields.split(', ')]
            else:
                fields = ''

            response = self.conn.call('RFC_READ_TABLE',
                                      QUERY_TABLE=table_name,
                                      DELIMITER=delimiter,
                                      OPTIONS=options,  # SQL WHERE
                                      FIELDS=fields)

            result = pd.DataFrame([[x.strip() for x in response['DATA'][n]['WA'].split(delimiter)] for n in range(len(response['DATA']))],
                                  columns=[x['FIELDNAME'] for x in response['FIELDS']])

            return result

        except (ABAPApplicationError, ABAPRuntimeError):
            print("An error occurred.")

            raise

    # Report generation example. You can make this kind of function of your own.
    def customer_master(self, company_codes):
        opt = "' OR BUKRS = '".join(company_codes)
        options = "(BUKRS = '" + opt + "') AND (LOEVM <> 'X')"

        knb1 = self.table_access("KNB1", options, "KUNNR")

        opt = "' OR KUNNR = '".join(knb1['KUNNR'].unique().tolist())
        options = "(KUNNR = '" + opt + "')"

        return self.table_access("KNA1", options, "KUNNR, LAND1, NAME1, NAME2, ORT01, ORT02, STRAS, TELF1, TELFX, BRSCH, NIELS, LZONE")