Welcome to Celoxtractor!

Celoxtractor is a Python package designed to let you develop your own Celonis IBC Extractor easily. The package empowers you to connect your Celonis IBC team to source systems in addition to those which are already supported by Celonis’ many out-of-the-box Extractors. Here are just a few of Celoxtractor features that you will benefit from:

• Complete data control: Python Extractors are running on your server with Celonis uplink technology. You have complete control over your data.

• Secured credentials and easy updates: you can flexibly define parameters that will be passed to your Python Extractor. No credentials will be stored in the script itself.

• Extensive functionality: you benefit from all IBC extraction capabilities, such as column selection, primary key override, pseudonymization or filtering.

• Full flexibility: Celoxtractor lets you add more and more data from your data source as you go. Also, you can set up delta loads by leveraging dynamic variables in your filters.

Getting Started

Install the latest version of Celoxtractor via:

pip install --extra-index-url=https://pypi.celonis.cloud/ celoxtractor

Then, just implement a Python class with two methods (see sub-pages of that documentation for more details):

# Import the required modules
from celoxtractor.extractor import CelonisExtractor
from celoxtractor.types import Table, Column, ExtractionData
from datetime import datetime

# Set up a class that extends from the Celoxtractor module's CelonisExtractor class
class MyExtractor(CelonisExtractor):

    # The get_metadata method defines the metadata for your data source
    def get_metadata(self):
        my_company_events = Table("my_company_event_logs")
        my_company_events.columns([Column("my_case_id", "INTEGER"), Column("my_activity", "STRING"), Column("my_timestamp", "DATETIME")])
        return [my_company_events]

    # The get_data_for_table method fetches data from your data source
    def get_data_for_table(self, table_name, parameters, is_delta_load, extraction_filter):
        extraction_data = ExtractionData()
        response = requests.get('https://my-company.com/internal/api/get_event_logs').json()
        for record in response:
            extraction_data.records.append([record["my_case_id"], record["my_activity"], datetime.strptime(record["my_timestamp"], '%Y-%m-%d %H:%M:%S')])
        return extraction_data

    # (You only have to define the methods not execute them.)

After having installed Celoxtractor and set up that class, you can directly leverage it to establish a new data connection between your data source and Celonis IBC team in no time. The easiest way to learn how Celoxtractor works is trying it out yourself! The Main Concepts section of that documentation will guide you through the required steps/methods and provide examples.

Join the Community

Ask questions, provide feedback and share best practices about Celoxtractor in the Celonis Community Forum here: https://community.celonis.com/search?q=python. We are looking forward to hearing from you!