Connecting to Data Sources: Databases, Flat Files, and Cloud Storage for Pipeline Ingestion

Every data pipeline starts with a single, unavoidable challenge: getting your hands on the data in the first place. Before you can transform, model, or analyze anything, you need to reach into the systems where data lives — a PostgreSQL database tracking customer orders, a CSV file exported from a legacy accounting system, an S3 bucket full of JSON logs from a mobile app — and pull that data into your pipeline reliably.

This sounds straightforward until you're staring at a connection error at 11 PM before a product launch, or you realize the CSV your vendor sends every Monday has been silently changing its column structure. The way you connect to your data sources is not just a technical detail — it determines how fragile or resilient your entire pipeline will be.

By the end of this lesson, you'll know how to connect to the three most common categories of data sources you'll encounter as a data engineer: relational databases, flat files, and cloud object storage. You'll understand not just the mechanics, but the reasoning — why each connection type works differently, what can go wrong, and how professionals handle these situations in production.

What you'll learn:

• How relational databases expose data through connection strings and SQL, and how to query them safely from Python
• How to read flat files (CSV, JSON, Parquet) in a pipeline-ready way that handles real-world messiness
• How to connect to cloud object storage on AWS S3, and how that differs from local file access
• How to manage credentials securely so you're not hardcoding passwords into your scripts
• How to structure your ingestion code so it's reusable and testable

---

Prerequisites

• Basic Python familiarity (you know what a function is, you've used pip install)
• A terminal or command prompt you're comfortable running commands in
• No database or cloud experience required — we'll explain everything from scratch

---

Understanding Data Sources: The Three Categories

Before we write a single line of code, let's build a mental model of what we're dealing with.

Think of a data source as a warehouse where information is stored. Different warehouses have different rules for how you get in and what you can take. A relational database is like a highly organized warehouse with a guard at the door — you need credentials to enter, and you request items through a formal query language (SQL). A flat file is like a filing cabinet someone left in a shared room — anyone with access to the room can open it, but the cabinet itself has no intelligence about what's inside. Cloud object storage is like a warehouse that lives off-site, managed by someone else — you need an account, a key, and you retrieve items by name over the internet.

These aren't just metaphors. They directly explain why the code for each connection type looks different, and why each has its own failure modes.

---

Connecting to Relational Databases

What a Database Connection Actually Is

A relational database like PostgreSQL, MySQL, or SQL Server is a running server process — a program that's constantly listening for requests. When your Python script "connects" to a database, it's opening a network socket (a communication channel) to that server, authenticating with a username and password, and then keeping that channel open to send SQL queries back and forth.

The key piece of information that describes how to reach a database is called a connection string (sometimes called a DSN — Data Source Name). A connection string is a single string of text that encodes everything needed to connect: the database type, the host address, the port, the database name, and the credentials. It looks like this:

postgresql://analytics_user:s3cur3p@$$w0rd@db.company.internal:5432/sales_db

Breaking this down:

• postgresql:// — the database driver to use
• analytics_user:s3cur3p@$$w0rd — username and password
• db.company.internal — the hostname where the server is running
• 5432 — the port (PostgreSQL's default)
• sales_db — the specific database to connect to

Setting Up: SQLAlchemy and psycopg2

In Python, the standard way to connect to relational databases is through SQLAlchemy, a library that provides a consistent interface regardless of which database you're using underneath. SQLAlchemy doesn't talk to databases itself — it uses database-specific drivers to do the actual communication. For PostgreSQL, that driver is psycopg2.

pip install sqlalchemy psycopg2-binary pandas

Your First Database Query in a Pipeline

Here's how a real ingestion function looks. We're pulling customer orders placed in the last 24 hours from a PostgreSQL database:

from sqlalchemy import create_engine, text
import pandas as pd
from datetime import datetime, timedelta

def ingest_recent_orders(connection_string: str, hours_back: int = 24) -> pd.DataFrame:
    """
    Pull orders from the last N hours from the sales database.
    Returns a DataFrame ready for downstream processing.
    """
    engine = create_engine(connection_string)
    
    cutoff_time = datetime.utcnow() - timedelta(hours=hours_back)
    
    query = text("""
        SELECT 
            order_id,
            customer_id,
            order_total,
            status,
            created_at
        FROM orders
        WHERE created_at >= :cutoff
        ORDER BY created_at DESC
    """)
    
    with engine.connect() as connection:
        result = pd.read_sql(query, connection, params={"cutoff": cutoff_time})
    
    return result


if __name__ == "__main__":
    conn_str = "postgresql://analytics_user:password@localhost:5432/sales_db"
    orders_df = ingest_recent_orders(conn_str)
    print(f"Ingested {len(orders_df)} orders")
    print(orders_df.head())

There are several important practices embedded in this code worth calling out:

Parameterized queries — notice the :cutoff placeholder in the SQL, and the params={"cutoff": cutoff_time} argument. We never build the SQL string by concatenating variables directly. Doing so creates a SQL injection vulnerability, where malicious input can alter the query itself. Always use parameters.

Context manager (with statement) — using with engine.connect() as connection: ensures the connection is properly closed even if an error occurs. Leaving database connections open is a common resource leak in pipeline code.

Returning a DataFrame — this function does one thing: get data and return it as a DataFrame. The transformation logic lives elsewhere. This separation makes the function easy to test.

Connection Pooling: Why Pipelines Don't Reconnect Every Time

If your pipeline runs a query every few minutes, opening and closing a fresh TCP connection each time is wasteful — establishing a database connection has real overhead (sometimes 20–100ms). SQLAlchemy handles this automatically through connection pooling: it maintains a small set of open connections and reuses them across queries. You don't have to configure this for basic use, but it's good to know it's happening under the hood.

---

Connecting to Flat Files

What Makes Flat Files Tricky

Flat files — CSVs, JSON files, Parquet files, Excel spreadsheets — seem like the simplest data source. No server to connect to, no credentials, just open a file and read it. In practice, flat files are where pipelines go to die quietly.

The problem is that flat files have no schema enforcement. A CSV file is just text. Nothing stops a vendor from adding a column, changing a column name, or switching from comma-separated to semicolon-separated between Monday's delivery and next Monday's. Databases have referential integrity and data types built in. Flat files have none of that.

Professional ingestion code for flat files includes explicit validation, not just reading.

Reading CSV Files Robustly

Here's a naive CSV read — the kind most tutorials show:

import pandas as pd
df = pd.read_csv("sales_data.csv")

And here's what happens in production when the vendor changes the delimiter to a semicolon, the encoding to Latin-1 (common in European countries), or adds a comment row at the top: your pipeline silently produces garbage data, or crashes, and you don't find out until someone notices the sales dashboard looks wrong.

Here's a more defensive ingestion function:

import pandas as pd
from pathlib import Path

EXPECTED_COLUMNS = {"transaction_id", "amount", "currency", "transaction_date", "merchant_name"}

def ingest_transaction_csv(file_path: str) -> pd.DataFrame:
    """
    Ingest a vendor transaction CSV with explicit validation.
    Raises ValueError if the file doesn't match the expected structure.
    """
    path = Path(file_path)
    
    if not path.exists():
        raise FileNotFoundError(f"Transaction file not found: {file_path}")
    
    if path.stat().st_size == 0:
        raise ValueError(f"Transaction file is empty: {file_path}")
    
    df = pd.read_csv(
        file_path,
        encoding="utf-8",
        delimiter=",",
        parse_dates=["transaction_date"],
        dtype={
            "transaction_id": str,
            "amount": float,
            "currency": str,
            "merchant_name": str,
        }
    )
    
    actual_columns = set(df.columns)
    missing_columns = EXPECTED_COLUMNS - actual_columns
    if missing_columns:
        raise ValueError(
            f"CSV is missing expected columns: {missing_columns}. "
            f"Got columns: {actual_columns}"
        )
    
    print(f"Successfully ingested {len(df)} rows from {path.name}")
    return df

Notice how we check for existence, check for empty files, specify encoding and delimiter explicitly, and validate the schema. This function will loudly fail with a clear error message instead of silently producing wrong data.

JSON Files: Nested Data Needs Special Handling

JSON is common for API exports and log data. The catch is that JSON is often nested — objects inside objects — while DataFrames are flat tables. Pandas read_json handles simple cases, but for nested JSON you'll often need json_normalize.

import json
import pandas as pd

def ingest_api_export(file_path: str) -> pd.DataFrame:
    """
    Ingest a nested JSON export from a CRM API.
    Each record has a top-level customer object with a nested 'address' object.
    """
    with open(file_path, "r", encoding="utf-8") as f:
        raw_data = json.load(f)
    
    # raw_data looks like:
    # [{"customer_id": "C001", "name": "Acme Corp", "address": {"city": "Austin", "state": "TX"}}, ...]
    
    df = pd.json_normalize(
        raw_data,
        sep="_"   # nested key "address.city" becomes column "address_city"
    )
    
    return df

Parquet: The Format You Should Be Using for Large Data

If you're dealing with files larger than a few hundred MB, Parquet is worth knowing about. Unlike CSV (plain text), Parquet is a columnar binary format — it compresses much better, stores data types explicitly, and reads far faster when you only need a few columns. A 1GB CSV often becomes a 150MB Parquet file that reads 5x faster.

pip install pyarrow

import pandas as pd

def ingest_parquet_events(file_path: str, columns: list = None) -> pd.DataFrame:
    """
    Ingest event data from a Parquet file.
    Use 'columns' to read only the fields you need — Parquet's big efficiency win.
    """
    df = pd.read_parquet(file_path, columns=columns)
    return df


# Only read 3 of potentially 50 columns — massive performance win
events = ingest_parquet_events(
    "app_events_2024_03.parquet",
    columns=["event_type", "user_id", "timestamp"]
)

Tip: When you're designing a pipeline that will pass data between stages stored as files, default to Parquet. Your future self will thank you when a CSV that would have taken 4 minutes to read takes 20 seconds.

---

Connecting to Cloud Object Storage

What Cloud Object Storage Actually Is

Services like AWS S3, Google Cloud Storage, and Azure Blob Storage are not file systems in the traditional sense. They're object stores — giant key/value stores where each "key" is a path-like string (e.g., data/raw/transactions/2024/03/15/batch_001.csv) and the "value" is the file's content. There's no real directory hierarchy — those slashes are just part of the key name. This distinction matters when you're listing "folders" programmatically.

You access cloud storage over HTTPS using APIs, not file system calls. This means every read and write involves a network round trip. It also means you need credentials — specifically, an access key ID and a secret access key (for AWS) — to prove you're authorized to access the bucket.

Setting Up AWS S3 Access with boto3

boto3 is the official Python library for AWS. It handles authentication, request signing, and all the AWS API details.

pip install boto3

The cleanest way to handle credentials is through environment variables or an AWS credentials file — never hardcoded in your script.

Set up credentials in your terminal (or in your environment configuration):

export AWS_ACCESS_KEY_ID="AKIAIOSFODNN7EXAMPLE"
export AWS_SECRET_ACCESS_KEY="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
export AWS_DEFAULT_REGION="us-east-1"

When boto3 initializes, it automatically looks for these environment variables. This keeps credentials out of your code entirely.

Listing and Reading Files from S3

import boto3
import pandas as pd
from io import BytesIO

def list_s3_files(bucket: str, prefix: str, suffix: str = ".csv") -> list:
    """
    List all files in an S3 'folder' (prefix) matching a suffix.
    Returns a list of full S3 keys.
    """
    s3_client = boto3.client("s3")
    
    response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)
    
    if "Contents" not in response:
        print(f"No files found in s3://{bucket}/{prefix}")
        return []
    
    keys = [
        obj["Key"] 
        for obj in response["Contents"] 
        if obj["Key"].endswith(suffix)
    ]
    
    return keys


def read_csv_from_s3(bucket: str, key: str) -> pd.DataFrame:
    """
    Read a single CSV file directly from S3 into a DataFrame.
    No need to download to disk first.
    """
    s3_client = boto3.client("s3")
    
    response = s3_client.get_object(Bucket=bucket, Key=key)
    
    # response["Body"] is a streaming object — we wrap it in BytesIO
    # so pandas can read it like a local file
    body = response["Body"].read()
    df = pd.read_csv(BytesIO(body))
    
    return df


def ingest_daily_s3_batch(bucket: str, date_prefix: str) -> pd.DataFrame:
    """
    Ingest all CSVs for a given date from S3, combine into one DataFrame.
    Example: ingest_daily_s3_batch("company-data-lake", "raw/transactions/2024/03/15/")
    """
    keys = list_s3_files(bucket, prefix=date_prefix, suffix=".csv")
    
    if not keys:
        raise ValueError(f"No CSV files found for prefix: {date_prefix}")
    
    dataframes = []
    for key in keys:
        print(f"Reading: s3://{bucket}/{key}")
        df = read_csv_from_s3(bucket, key)
        df["_source_file"] = key  # Track which file each row came from
        dataframes.append(df)
    
    combined = pd.concat(dataframes, ignore_index=True)
    print(f"Ingested {len(combined)} total rows from {len(keys)} files")
    return combined

The BytesIO wrapper deserves a moment of explanation. Pandas' read_csv expects something it can read like a file — it calls .read() methods on it. The S3 response body is a streaming object that also supports .read(), but BytesIO wraps the raw bytes in an interface that pandas is guaranteed to handle correctly across different versions.

Warning: If an S3 bucket has more than 1,000 objects, list_objects_v2 only returns the first 1,000. For large buckets, you need to use pagination — check the boto3 docs for paginate. This is a common, silent bug in beginner pipeline code.

---

Managing Credentials Securely

Here is the single most common security mistake in pipeline code:

# NEVER DO THIS
engine = create_engine("postgresql://admin:MyPassword123@prod-db.company.com:5432/main")

When this script gets committed to a Git repository — which happens constantly, even at large companies — those credentials are now in version control history forever, potentially exposed to anyone with repository access.

The professional approach has two main patterns:

Pattern 1: Environment Variables

import os
from sqlalchemy import create_engine

def get_db_engine():
    db_url = (
        f"postgresql://{os.environ['DB_USER']}:{os.environ['DB_PASSWORD']}"
        f"@{os.environ['DB_HOST']}:{os.environ.get('DB_PORT', '5432')}"
        f"/{os.environ['DB_NAME']}"
    )
    return create_engine(db_url)

Your pipeline script reads credentials from environment variables. On your local machine, you set these in your shell profile or in a .env file (which you add to .gitignore). In production, your infrastructure (Docker, Kubernetes, a CI/CD system) injects these values.

Pattern 2: A .env File with python-dotenv

pip install python-dotenv

Create a file named .env in your project root (and add it to .gitignore immediately):

DB_HOST=localhost
DB_PORT=5432
DB_NAME=sales_db
DB_USER=analytics_user
DB_PASSWORD=s3cur3p@$$w0rd

Then in your code:

from dotenv import load_dotenv
import os

load_dotenv()  # Reads .env file and sets environment variables

db_user = os.environ["DB_USER"]
db_password = os.environ["DB_PASSWORD"]

This pattern is excellent for local development and works seamlessly in team environments — everyone has their own .env file with their own credentials, and none of it ever touches the repository.

---

Hands-On Exercise

Let's put it all together. Your task is to build a small ingestion script that reads from two sources and combines them.

Scenario: You're building a pipeline for a retail analytics team. They have:

1. A CSV file (store_locations.csv) with columns: store_id, city, state, region
2. A SQLite database (sales.db) with a table daily_sales containing: store_id, sale_date, revenue

Your goal is to ingest both, join them, and output a combined DataFrame showing total revenue by region.

Step 1: Create the test data. Run this script once to generate the files:

import pandas as pd
import sqlite3

# Create CSV
stores = pd.DataFrame({
    "store_id": ["S001", "S002", "S003", "S004"],
    "city": ["Austin", "Denver", "Chicago", "Seattle"],
    "state": ["TX", "CO", "IL", "WA"],
    "region": ["South", "West", "Midwest", "West"]
})
stores.to_csv("store_locations.csv", index=False)

# Create SQLite database
conn = sqlite3.connect("sales.db")
sales = pd.DataFrame({
    "store_id": ["S001", "S002", "S003", "S004", "S001", "S003"],
    "sale_date": ["2024-03-15"] * 6,
    "revenue": [12400.50, 8900.00, 15200.75, 9800.25, 6700.00, 4300.50]
})
sales.to_sql("daily_sales", conn, if_exists="replace", index=False)
conn.close()

print("Test data created.")

Step 2: Write the ingestion script from scratch. Try to implement:

• A ingest_store_locations(file_path) function that reads the CSV with explicit dtype declarations
• A ingest_daily_sales(db_path, sale_date) function that queries the SQLite database for a specific date using a parameterized query
• A build_regional_summary(stores_df, sales_df) function that joins and aggregates the data

Step 3: Validate your output. The combined DataFrame should show that the "West" region has stores in Denver and Seattle, with combined revenue from both.

SQLite works exactly like PostgreSQL in terms of the SQLAlchemy interface — use create_engine(f"sqlite:///{db_path}") as the connection string.

---

Common Mistakes & Troubleshooting

"Could not connect to server: Connection refused" The database server isn't running, or the host/port is wrong. For local development, verify the service is started. For remote databases, confirm the host address and that the server's firewall allows connections from your IP.

"SSL connection required" or certificate errors Many cloud databases require SSL. Add ?sslmode=require to your PostgreSQL connection string: postgresql://user:pass@host:5432/db?sslmode=require.

CSV reads fine locally, fails in production with encoding errors Files from Windows systems often use Windows-1252 encoding. Files from older enterprise systems sometimes use ISO-8859-1 (Latin-1). Always specify encoding explicitly. When you don't know, try encoding="utf-8-sig" first (handles UTF-8 with Windows BOM), then encoding="latin-1" as a fallback.

S3 returns "NoCredentialsError" boto3 looks for credentials in this order: environment variables → ~/.aws/credentials file → IAM instance role. If none are found, you get this error. Double-check your environment variables are set in the same shell session where you're running the script.

DataFrame has unexpected columns after a vendor CSV update This is why we validate schema explicitly. Add logging to your ingestion functions that records column names on every run — when the pipeline breaks, you'll immediately see what changed.

pd.read_sql with SQLAlchemy 2.0 gives a warning about connections SQLAlchemy 2.0 changed how connections are passed to pandas. Always pass a connection object (from with engine.connect() as connection) rather than the engine directly to pd.read_sql.

---

Summary & Next Steps

You now have a working mental model and practical code patterns for the three major data source categories you'll encounter in real pipelines:

• Relational databases accessed via SQLAlchemy with parameterized queries, connection pooling, and proper connection lifecycle management
• Flat files (CSV, JSON, Parquet) ingested defensively with explicit schema validation, encoding specification, and format-appropriate tooling
• Cloud object storage (S3) accessed via boto3, reading files directly into memory without intermediate downloads, and understanding the pagination limitation

The security practices are just as important as the technical patterns: environment variables and .env files keep credentials out of your codebase, which is non-negotiable in any professional context.

Where to go next:

The natural next step in the Data Pipeline Fundamentals path is Data Transformation Patterns — once your data is ingested, you need to clean it, normalize it, and shape it for downstream consumers. The ingestion functions you wrote here produce DataFrames; the transformation layer picks up from there.

You should also explore incremental ingestion — rather than pulling all data every run, how do you track what you've already processed and only pull new records? This is the difference between a pipeline that runs in seconds and one that re-scans millions of rows every hour.

Finally, consider the observability side of ingestion: logging row counts, recording the timestamp of each run, alerting when a source file doesn't arrive. Ingestion is the stage most exposed to the outside world, and the outside world is unpredictable.