dltpersonio-source

父表

'employees', 
'absences', 
'absence_types', 
'attendances'

其中一些表有子表

要将父表与子表连接起来，请使用以下连接 parent.dlt_id = child.parent_dlt_id

用法

安装库

pipx安装dltpersonio-source 如果找不到库，请确保您具有根据pyproject.toml文件所需的Python版本。

按照以下方式运行源代码以加载示例数据集。

添加凭据并删除dummy_data标志以启用加载数据。

首先，导入加载方法并添加您的凭据

from dlt_personio_source import load_personio_tables

#target credentials
# example for bigquery
creds = {
  "type": "service_account",
  "project_id": "zinc-mantra-353207",
  "private_key_id": "example",
  "private_key": "",
  "client_email": "example@zinc-mantra-353207.iam.gserviceaccount.com",
  "client_id": "100909481823688180493"}
  
# or example for redshift:
# creds = ["redshift", "database_name", "schema_name", "user_name", "host", "password"]

#Personio credentials
#get credentials at this url - replace"test-1" with your org name
#https://test-1.personio.de/configuration/api/credentials/management
client_id = ''
client_secret = ''

然后，您可以使用以下代码进行序列加载

# remove some tables from this list of you only want some endpoints
tables = ['employees', 'absences', 'absence_types', 'attendances']
load_personio_tables(client_id=client_id,
                     client_secret=client_secret,
                     target_credentials=creds,
                     tables=tables,
                     schema_name='personio_raw',
                     dummy_data=True)

或者，对于并行加载，为每个表创建类似以下内容的airflow任务

tables = ['employees', 'absences', 'absence_types', 'attendances']
for table in tables:
    load_personio_tables(client_id='',
                         client_secret='',
                         target_credentials=creds,
                         tables = [table],
                         schema_name='personio_raw',
                         dummy_data = True)

如果您想自己构建管道或以不同的方式使用源

from dlt_personio_source import PersonioSource, PersonioSourceDummy

prod = PersonioSource(client_id='',
              client_secret='')
              
dummy = PersonioSourceDummy()

sample_data = dummy.tasks() 

for task in tasks:
    print(task['table_name'])
    for row in task['data']
        print(row)