> ## Content Index > Fetch the complete content index at: https://articles.akadata.co.uk/llms.txt > Use this file to discover other available public pages before exploring further. # Building and Automating a Data Pipeline with Python: A Practical Tutorial - URL: https://articles.akadata.co.uk/building-and-automating-a-data-pipeline-with-python-a-practical-tutorial/ - Published: 2025-02-22T22:24:50.000Z - Updated: 2025-02-23T00:47:41.000Z - Author: AKADATA ### Introduction In a world increasingly driven by data, having a robust pipeline to ingest, transform, and deliver information is essential. Whether integrating with an external API or consolidating data from multiple sources, an automated pipeline ensures you can reliably handle information at scale. This tutorial provides a hands-on guide to constructing a data pipeline in Python, including code examples and recommended practices for automation. ## Why Build a Data Pipeline? A data pipeline automates data flow from its source to a destination (such as a database, data warehouse, or data visualisation tool). This eliminates manual steps, reduces the risk of errors, and provides an auditable, repeatable process. By the end of this tutorial, you'll have a working pipeline that: 1. **Fetches Data** from a publicly available API. 2. **Transforms Data** to match your requirements. 3. **Stores Data** in a local or remote database. 4. **Automates the Workflow** using a scheduling tool or CI/CD platform. ## Prerequisites - **Python 3.8+** (or a more recent version) - **Basic knowledge of Python** (functions, virtual environments, pip installs, etc.) - **A local or remote database** (e.g., PostgreSQL, MySQL, or SQLite) - **Familiarity with Git** and a CI/CD tool (e.g., Jenkins, GitHub Actions, or GitLab CI) if you plan to automate. If you're new to Python or open-source tools, don't worry. This guide is designed to be approachable for beginners while offering insights that seasoned developers can appreciate. ## 1\. Designing the Pipeline Before writing code, let's define a small but realistic scenario. Suppose we need to: 1. **Pull Real-Time Exchange Rates** from a public API. 2. **Transform** the data to highlight specific currency pairs. 3. **Store** the results in a local PostgreSQL database for further analysis. 4. **Schedule** the entire Workflow to run daily at midnight. ### Data Flow Diagram ```text API (Exchange Rates) --> Python Script (Ingestion) --> Python Script (Transformation) --> PostgreSQL DB | Scheduled via CRON or CI/CD ``` ## 2\. Setting Up Your Environment Create a new directory for your pipeline project: ```bash mkdir currency-pipeline cd currency-pipeline ``` Then, initialise a Python virtual environment: ```bash python3 -m venv venv source venv/bin/activate # Linux/Mac # or .\venv\Scripts\activate on Windows ``` Install required libraries in a `requirements.txt` file: ```text requests pandas SQLAlchemy psycopg2 ``` Then run: ```bash pip install -r requirements.txt ``` - **Requests** for HTTP calls to the external API. - **Pandas** for data manipulation. - **SQLAlchemy** and **psycopg2** for interacting with PostgreSQL. ## 3\. Fetching Data from an API Below is a Python script called `fetch_data.py` that requests exchange rate data from a public API (e.g., "ExchangeRate-API" or any free alternative). Adjust the URL and parameters to suit your chosen API. ```python # fetch_data.py import requests import pandas as pd from datetime import datetime def fetch_exchange_rates(api_key: str, base_currency: str = "USD") -> pd.DataFrame: """ Fetches exchange rate data from a public API and returns a DataFrame. :param api_key: Your API key for the exchange rate service. :param base_currency: The currency to use as a base for exchange rates. :return: Pandas DataFrame with currency rates. """ url = f"https://v6.exchangerate-api.com/v6/{api_key}/latest/{base_currency}" response = requests.get(url) response.raise_for_status() # Raises HTTPError if status != 200 data = response.json() # data['conversion_rates'] is a dictionary e.g. {"GBP": 0.81, "EUR": 0.94, ...} rates = data.get("conversion_rates", {}) # Convert dictionary to DataFrame df = pd.DataFrame(list(rates.items()), columns=["Currency", "Rate"]) df["BaseCurrency"] = base_currency df["Timestamp"] = datetime.now() return df if __name__ == "__main__": # Example usage test_api_key = "YOUR_EXCHANGE_RATE_API_KEY" df_rates = fetch_exchange_rates(test_api_key, "USD") print(df_rates.head()) ``` ### Explanation - We call a public API endpoint with requests.get(). - The returned JSON data is parsed and placed into a Pandas DataFrame for easier manipulation. - We add columns like BaseCurrency and Timestamp to track metadata. ## 4\. Transforming Data Now, we create a transformation script (`transform_data.py`). Suppose we only care about a few currency pairs—say, USD to GBP, USD to EUR, and USD to JPY. ```python # transform_data.py import pandas as pd def filter_currencies(df: pd.DataFrame, currency_list: list) -> pd.DataFrame: """ Filters the DataFrame to only include specified currencies. :param df: Original DataFrame from fetch_data. :param currency_list: List of currency codes to keep. :return: Filtered DataFrame. """ filtered_df = df[df["Currency"].isin(currency_list)].copy() return filtered_df if __name__ == "__main__": # For testing sample_data = { "Currency": ["GBP", "EUR", "JPY", "INR", "CAD"], "Rate": [0.82, 0.94, 131.2, 82.3, 1.26], "BaseCurrency": ["USD"]*5, "Timestamp": ["2025-01-01 12:00"]*5 } df_sample = pd.DataFrame(sample_data) filtered = filter_currencies(df_sample, ["GBP", "EUR", "JPY"]) print(filtered) ``` ### Explanation - `filter_currencies` takes a DataFrame and returns rows matching a given currency code list. - Additional transformations could be included here, such as rounding decimal places or calculating percentage changes. ## 5\. Storing Data in PostgreSQL Next, let's create a script called `store_data.py` to push our filtered data into a PostgreSQL database. We'll use **SQLAlchemy** for convenience. ```python # store_data.py import pandas as pd from sqlalchemy import create_engine def store_to_postgres(df: pd.DataFrame, table_name: str, db_url: str): """ Stores the DataFrame to a specified table in PostgreSQL. :param df: DataFrame to store. :param table_name: Name of the table in PostgreSQL. :param db_url: Connection string for PostgreSQL. """ engine = create_engine(db_url) df.to_sql(table_name, engine, if_exists="append", index=False) print(f"Data successfully inserted into {table_name}.") if __name__ == "__main__": # Example usage sample_data = { "Currency": ["GBP", "EUR", "JPY"], "Rate": [0.82, 0.94, 131.2], "BaseCurrency": ["USD"]*3, "Timestamp": ["2025-01-01 12:00"]*3 } df_sample = pd.DataFrame(sample_data) # Replace with your actual PostgreSQL credentials postgres_url = "postgresql://user:password@localhost:5432/mydatabase" store_to_postgres(df_sample, "exchange_rates", postgres_url) ``` ### Explanation - `create_engine` from SQLAlchemy is used to connect to PostgreSQL via a URL of the form `postgresql://user:password@hostname:port/dbname`. - `df.to_sql` handles the insertion. We set `if_exists= "append"` to add new rows to the existing table. ## 6\. Integrating the Pipeline Finally, let's combine everything in a single script (`pipeline.py`) that calls each step in the sequence: ```python # pipeline.py import os import sys from fetch_data import fetch_exchange_rates from transform_data import filter_currencies from store_data import store_to_postgres def run_pipeline(): # Load environment variables or config api_key = os.getenv("EXCHANGE_API_KEY", "YOUR_EXCHANGE_RATE_API_KEY") db_url = os.getenv("DB_URL", "postgresql://user:password@localhost:5432/mydatabase") table_name = "exchange_rates" # 1. Fetch Data df_fetched = fetch_exchange_rates(api_key, "USD") # 2. Transform Data df_transformed = filter_currencies(df_fetched, ["GBP", "EUR", "JPY"]) # 3. Store Data store_to_postgres(df_transformed, table_name, db_url) if __name__ == "__main__": run_pipeline() ``` ### Explanation - The script uses environment variables to manage sensitive credentials like API keys and database URLs. - We fetch the data, transform it, and store it in PostgreSQL. ## 7\. Automating the Pipeline ### Option A: Cron Job on Linux For a simple daily schedule, you can use a CRON job: ```bash crontab -e ``` Then add: ```bash 0 0 * * * /path/to/pipeline/venv/bin/python /path/to/pipeline/pipeline.py >> /path/to/logs/pipeline.log 2>&1 ``` This runs `pipeline.py` at midnight daily and logs output to `pipeline.log`. ### Option B: CI/CD (Jenkins Example) Below is a Jenkinsfile that automatically runs the pipeline whenever you push changes to your repository: ```groovy pipeline { agent any stages { stage('Checkout') { steps { git url: 'https://github.com/YourUser/currency-pipeline.git', branch: 'main' } } stage('Install') { steps { sh 'pip install -r requirements.txt' } } stage('Run Pipeline') { steps { sh 'python pipeline.py' } } } post { success { echo 'Pipeline executed successfully!' } failure { echo 'Pipeline failed. Check logs for details.' } } } ``` ### Option C: GitHub Actions For GitHub Actions, create a `.github/workflows/pipeline.yml`: ```yaml name: Data Pipeline on: push: branches: [ "main" ] schedule: - cron: '0 0 * * *' jobs: build: runs-on: ubuntu-latest steps: - name: Checkout uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.9' - name: Install Dependencies run: | pip install --upgrade pip pip install -r requirements.txt - name: Run Pipeline run: python pipeline.py ``` This file triggers your pipeline whenever you push to the `main` branch and once daily at midnight. ## 8\. Best Practices and Further Considerations - **Error Handling:** Implement try-except blocks or robust error logging to gracefully handle API downtime or database connectivity issues. - **Version Control for Data:** If you need data lineage or auditing, consider storing pipeline outputs in a versioned data store or applying tags. - **Security:** Avoid hardcoding credentials. Use environment variables or secret managers (e.g., HashiCorp Vault, AWS Secrets Manager) for sensitive information. - **Scalability:** Consider distributed computing frameworks (e.g., Apache Spark) for massive datasets. Alternatively, containerise your pipeline and deploy it on Kubernetes to handle large loads. ## Conclusion Building and automating a data pipeline doesn't have to be overly complicated. You gain a reliable and repeatable workflow by breaking the process into distinct stages—fetching, transforming, and storing—and then scheduling these tasks. Python's rich ecosystem of libraries, combined with open-source CI/CD tools, makes it easier than ever to integrate data from external APIs, perform transformations, and deposit the final product into a database or data warehouse. These principles remain the same whether you're managing small-scale analytics or orchestrating massive data workloads. With robust automation, version control, and a clear understanding of each step, you can streamline your data processes and focus on deriving insights rather than wrestling with ad-hoc scripts. Feel free to adapt and expand the code examples to suit your needs. **Need Assistance?** If you're looking to refine or scale your data pipelines or if you'd like more advanced tutorials, let me know! I can discuss container orchestration, distributed computing, advanced scheduling, and more. Good luck with your pipeline building and automation journey!