Designing a Python Data Collection and Automation Pipeline

When first organizing Python, grammar items such as variables, data types, control statements, and functions come first. But where Python becomes particularly useful from a backend and platform perspective is that “iterative data operations can be created into small pipelines.” The flow of importing external data through HTTP, extracting necessary fields from JSON or HTML, organizing them with Pandas, and then passing them to a file, database, or API response leads to a short code.

Web crawling, Open API, data analysis, and Jupyter Notebook practice are not separate topics. In hands-on work, it is connected to the flow of collecting operational indicators, importing external public data, organizing news and documents, or creating batch reports. This article is a record of the basic Python learning contents regrouped into elements that make up an automation pipeline, rather than a grammar list.

The execution environment must be reproducible

Python is easy to install, but as the number of projects increases, environments mix easily. Installing all libraries in global Python makes it difficult to keep track of which projects require which versions of packages. So, even if it is a small practice, create a virtual environment first.

python -m venv .venv

# Windows PowerShell
.\.venv\Scripts\Activate.ps1

# macOS / Linux
source .venv/bin/activate

python -m pip install --upgrade pip
python -m pip install requests beautifulsoup4 pandas pyarrow openpyxl jupyter
python -m pip freeze > requirements.txt

It is easy to check whether the currently active Python executable and pip are in the same environment by using python -m pip instead of pip install. This is especially useful in Windows environments where multiple Python versions are installed.

The project folder separates code and data. In the practice stage, a simple structure is sufficient.

project/
  code/
    collect_market_data.py
    transform_prices.py
  data/
    raw/
    processed/
  requirements.txt

Jupyter Notebooks are great for exploration and visualization, but they are weak as the final form of operational automation. It is a good idea to move code tested on a laptop into reusable functions and scripts. A laptop is close to a workbench where you can quickly see “what has been checked,” and it is safer to manage the code to be executed in the scheduler or CI as a .py file.

Python basics are a tool for expressing data flow

Python's basic data types serve as data models in automation code. The JSON object of the API response is easy to handle with dict and list, and the processing results of multiple stages can be divided into functions and connected.

def normalize_price(value: str) -> int:
    return int(value.replace(",", "").strip())

def build_record(name: str, price: str, source: str) -> dict:
    return {
        "name": name.strip(),
        "price": normalize_price(price),
        "source": source,
    }

Dynamic typing is great for writing code quickly, but in a data pipeline, even a slight change in the input format can reveal errors late. Therefore, type hints are added to function boundaries, and external inputs are explicitly converted.

from dataclasses import dataclass

@dataclass(frozen=True)
class MarketItem:
    name: str
    price: int
    source: str

def parse_item(row: dict) -> MarketItem:
    return MarketItem(
        name=str(row["name"]).strip(),
        price=int(row["price"]),
        source=str(row.get("source", "unknown")),
    )

Lists, tuples, sets, and dictionaries each have different uses. Lists are natural when processing lists of records while maintaining order. Sets are convenient for removing duplicates, and dictionaries are good for searching based on identifiers. Tuples are lightweight when expressing immutable coordinates or pairs.

symbols = ["005930", "000660", "035420"]
unique_symbols = set(symbols)
symbol_to_name = {
    "005930": "Samsung Electronics",
    "000660": "SK hynix",
}

Appropriate use of these basic data structures can greatly reduce the complexity of small batch programs.

HTTP requests are made with failure as the default value.

Both web crawling and Open API calls start with HTTP requests. The URL consists of scheme, domain, path, and query string, and the response has a status code and body. 2xx is success, 3xx is redirect, 4xx is client error, and 5xx is server error. Automation code must check this state.

import requests

def fetch_json(url: str, params: dict | None = None) -> dict:
    response = requests.get(
        url,
        params=params,
        headers={"User-Agent": "tykdev-data-pipeline/1.0"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

If timeout is not specified, the process may hang for a long time when the external server does not respond. If raise_for_status() is not called, a 404 or 500 response may be processed as a normal body, but a strange error may occur at a later stage.

In crawling, you must first check the policy of the collection target. You should look at robots.txt, terms of service, request frequency limits, copyright, and whether personal information is processed. Just because you can retrieve visible data from your browser doesn't mean automatic collection is always a good idea.

from urllib.robotparser import RobotFileParser

def can_fetch(base_url: str, target_url: str) -> bool:
    parser = RobotFileParser()
    parser.set_url(f"{base_url}/robots.txt")
    parser.read()
    return parser.can_fetch("tykdev-data-pipeline", target_url)

If data is provided by API, API is given priority over HTML scraping. The API has a clear structure, authentication and quota policies are documented, and field changeability is relatively easy to manage.

HTML parsing must be able to tolerate changes to the DOM structure

HTML scraping begins by looking at the Network and Elements tabs of the developer tools together. You must first distinguish whether the page is downloaded as HTML from a server or rendered by calling an API using JavaScript in the browser. If the actual data comes down to a separate JSON API, it may be more reliable to reproduce the API request rather than parse the HTML.

When using BeautifulSoup, do not focus too deeply on CSS selectors. Structure-sensitive selectors such as body > div:nth-child(3) > table > tr... break even with small markup changes. If possible, use meaningful classes, ids, table headers, and link attributes as standards.

from bs4 import BeautifulSoup

def parse_news(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    items = []

    for link in soup.select("a.news-title"):
        title = link.get_text(strip=True)
        href = link.get("href")

        if not title or not href:
            continue

        items.append({
            "title": title,
            "url": href,
        })

    return items

At the HTML parsing stage, the fact that there may be no data is handled as a normal flow. Whether the selector should fail immediately when it finds nothing, store an empty result, or send a notification depends on the purpose of the pipeline. If you are collecting operational metrics, empty results may be a sign of failure, or if you are gathering news, there may be few results at certain times.

Pandas is strong in preprocessing and verification

Pandas is good at handling tabular data. You can read CSV, Excel, and JSON, and quickly perform missing value processing, data type conversion, filtering, grouping, and joining. In the lab, functions such as Series, DataFrame, loc, iloc, groupby, merge, read_csv, and to_csv were confirmed.

import pandas as pd

def normalize_items(items: list[dict]) -> pd.DataFrame:
    df = pd.DataFrame(items)

    if df.empty:
        return pd.DataFrame(columns=["name", "price", "source", "collected_at"])

    df["name"] = df["name"].astype(str).str.strip()
    df["price"] = (
        df["price"]
        .astype(str)
        .str.replace(",", "", regex=False)
        .astype(int)
    )
    df["collected_at"] = pd.Timestamp.utcnow()
    return df.drop_duplicates(subset=["name", "source"])

The important thing in Pandas is to verify the conversion results not just visually, but also with code. For example, the price cannot be negative, and required columns cannot be empty.

def validate_frame(df: pd.DataFrame) -> None:
    required = {"name", "price", "source"}
    missing = required - set(df.columns)

    if missing:
        raise ValueError(f"missing columns: {sorted(missing)}")

    if df["name"].isna().any():
        raise ValueError("name contains null")

    if (df["price"] < 0).any():
        raise ValueError("price contains negative value")

Excel is strong for quick manual tasks and visual verification, but has limitations for large-scale, repeatable processing and versioning. Python requires initial environment setup, but because data collection and preprocessing procedures are left as code, it is easy to re-execute and review the same work.

Connect with a small pipeline

Dividing collection, parsing, purification, verification, and storage into functions makes the flow clearer.

from pathlib import Path

DATA_DIR = Path("data")

def run_pipeline() -> None:
    html = requests.get(
        "https://example.com/news",
        headers={"User-Agent": "tykdev-data-pipeline/1.0"},
        timeout=10,
    ).text

    records = parse_news(html)
    df = pd.DataFrame(records)
    validate_news_frame(df)

    output = DATA_DIR / "processed" / "news.parquet"
    output.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(output, index=False)

def validate_news_frame(df: pd.DataFrame) -> None:
    if "title" not in df.columns or "url" not in df.columns:
        raise ValueError("news frame requires title and url")

if __name__ == "__main__":
    run_pipeline()

At first, saving it as a local file will suffice. If scheduling becomes necessary later, it can be moved to an execution environment such as GitHub Actions, cron, Azure Functions, or AWS Lambda. The storage destination can also be changed from CSV to Parquet, PostgreSQL, Blob Storage, S3, Cosmos DB, etc. The important thing is that the function boundaries are separated, so even if the collection source and repository change, the entire code is not changed.

Automation pipelines require operational considerations even though their code is short. External API keys are separated into environment variables or secret stores, and a retry policy is set for request failure. If the collection time is long, timeout and pagination must be divided, and an idempotency key or unique constraint must be placed to prevent duplicate storage. It is easy to track it later by leaving metadata such as collection time, source URL, and version in the result file.

import os

API_KEY = os.environ["OPEN_DATA_API_KEY"]

params = {
    "serviceKey": API_KEY,
    "pageNo": 1,
    "numOfRows": 100,
}

Jupyter is searched, script is set as execution path

Jupyter Notebook is great for checking the shape of data, drawing graphs, and quickly experimenting with preprocessing ideas. You can quickly identify data characteristics by using tools such as %timeit, %whos, describe(), and value_counts().

df.info()
df.describe()
df["source"].value_counts()

However, in laptops, the cell execution order can easily be distorted, and hidden states can remain. Therefore, it is better to move tasks that need to be executed repeatedly into scripts, and keep the notebook close to the document explaining the analysis results. In team work, requirements.txt, sample data, execution commands, and output location must be recorded together to ensure reproducibility.

Cleanup

The starting point for learning Python is the syntax, but the real power comes from creating repeatable automated flows. Variables and data types represent external data, functions divide the collection and purification steps, requests and BeautifulSoup handle HTTP and HTML, and Pandas converts tabular data into a verifiable structure. Jupyter is good for navigation, and script execution and deployment.

If you create an automation pipeline, no matter how small, to the end, it will be directly connected to the backend and platform work. API calls require timeouts and authentication, collected data requires schema validation, and storage requires redundancy and failure recovery strategies. Python is a good language to quickly experiment with this flow, verify it, and then turn it into a small, operable tool.