meltano_project_init_and_elt_pipeline_execution.py

1import os
2from meltano.core.project import Project
3from meltano.core.project_init_service import ProjectInitService
4from meltano.core.elt_context import ELTContextBuilder
5from meltano.core.elt_run_service import ELTRunService
6from meltano.core.logging import setup_logging
7
8# Set up basic logging
9setup_logging()
10
11# 1. Initialize a Meltano project in a specific directory
12project_path = os.path.join(os.getcwd(), "my_meltano_project")
13init_service = ProjectInitService(project_path)
14project = init_service.init()
15
16# 2. Define the pipeline components (Extractor and Loader)
17# Note: These must be installed in the meltano.yml or added via project.plugins
18extractor = "tap-gitlab"
19loader = "target-jsonl"
20job_id = "gitlab-to-jsonl"
21
22# 3. Create the ELT Context
23context_builder = ELTContextBuilder(
24    project,
25    extractor=extractor,
26    loader=loader,
27    job_id=job_id
28)
29elt_context = context_builder.context()
30
31# 4. Run the ELT pipeline
32run_service = ELTRunService(elt_context)
33run_service.run()
34
35print(f"Pipeline {job_id} completed successfully.")