dbpipe is a lightweight and simple way to manage data pipelines.
graph LR
A(Endpoints)-->B(Pipes)
B-->C(Jobs)
D(Schedules)-->C
C-.->E(Clusters)
from dbpipe import EndPoint
facebook = EndPoint('Facebook','API','https://facebook.com/Posts')
facebook{'name': 'Facebook', 'type': 'API', 'location': 'https://facebook.com/Posts'}
facebook.save()posttable = EndPoint('DW.Facebook.Posts','Database','ServerName')
posttable{'name': 'DW.Facebook.Posts', 'type': 'Database', 'location': 'ServerName'}
posttable.save()from dbpipe import Pipe
pipe = Pipe(
name='DW',
sources=[facebook],
destination=posttable,
processfile="Test.py"
)
pipe{'name': 'DW', 'sources': [{'name': 'Facebook', 'type': 'API', 'location': 'https://facebook.com/Posts'}], 'destination': {'name': 'DW.Facebook.Posts', 'type': 'Database', 'location': 'ServerName'}, 'logfile': None, 'processfile': 'Test.py'}
pipe.to_dict(){'name': 'DW',
'sources': [{'name': 'Facebook',
'type': 'API',
'location': 'https://facebook.com/Posts'}],
'destination': {'name': 'DW.Facebook.Posts',
'type': 'Database',
'location': 'ServerName'},
'logfile': None,
'processfile': 'Test.py'}
pipe.save()from dbpipe import Schedule
schedule = Schedule(frequency="Daily", start_time="8:00AM")
schedule{'frequency': 'Daily', 'start_time': '8:00AM', 'end_time': None, 'time_zone': 'UTC'}
schedule.to_dict(){'frequency': 'Daily',
'start_time': '8:00AM',
'end_time': None,
'time_zone': 'UTC'}
from dbpipe.core.pipes import Cluster
clstr = Cluster([pipe,pipe])
clstr[{'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}, {'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}]
from dbpipe import Job
job = Job('My Job',schedule=schedule,jobs=clstr)
job{'name': 'My Job', 'schedule': {'frequency': 'Daily', 'start_time': '8:00AM', 'end_time': None, 'time_zone': 'UTC'}, 'jobs': [{'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}, {'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}]}
job.save()from dbpipe import read_pipe
pipe = read_pipe('pipes/DW.json')
pipe{'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}
pipe.to_dict(){'name': 'DW',
'sources': ['AdSpend', 'SocialStats'],
'destination': 'DW',
'logfile': None,
'processfile': 'Test.py'}
from dbpipe import read_job
job = read_job('jobs/My Job.json')
job{'name': 'My Job', 'schedule': {'frequency': 'Daily', 'start_time': '8:00AM', 'end_time': None, 'time_zone': 'UTC'}, 'jobs': [{'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}, {'name': 'DW', 'sources': ['AdSpend', 'SocialStats'], 'destination': 'DW', 'logfile': None, 'processfile': 'Test.py'}]}
job.to_dict(){'name': 'My Job',
'schedule': {'frequency': 'Daily',
'start_time': '8:00AM',
'end_time': None,
'time_zone': 'UTC'},
'jobs': [{'name': 'DW',
'sources': ['AdSpend', 'SocialStats'],
'destination': 'DW',
'logfile': None,
'processfile': 'Test.py'},
{'name': 'DW',
'sources': ['AdSpend', 'SocialStats'],
'destination': 'DW',
'logfile': None,
'processfile': 'Test.py'}]}
from dbpipe.lineage.mermaid import generate_mermaid_markdown_file
generate_mermaid_markdown_file('pipes','test.md')
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent