In the last part of this series I explored the CoinMarketCap API. In this part i’ll be extending my pydantic API with a new domain to handle the data. While I could’ve gone the boring route of shoving CoinMarketCap data straight into my Postgres database using an Airflow DAG, I opted for the cooler, more maintainable, and yes, slightly over-engineered option: building an API layer.

Why, you ask? Well, because APIs aren’t just for making developers look smart at conferences. They’re functional, versatile, and provide a host of benefits that I’m going to gleefully outline before diving into the process of extending my FastAPI app to handle cryptocurrency data.

Why Use an API Instead of Direct Database Inserts?

Sure, you could brute-force your data into a warehouse. But here’s why using an API is like choosing a Swiss Army knife over a plain butter knife:

  1. Centralized Database Operations 🛠️
    Encapsulating all database logic within the API ensures that DAGs don’t turn into a spaghetti code nightmare. I prefer one place for all CRUD operations.

  2. Self-Documenting Models and Schemas đź“ś
    With FastAPI and Pydantic, you’re essentially creating API documentation as you build. Who doesn’t love killing two birds with one stone (metaphorically, of course)?

  3. Abstraction Layers Are Your Friend 🧩
    Want to refactor the database schema? Switch to a different database? With an API, your external systems remain blissfully unaware. Your API does the heavy lifting.

  4. Version Control 🗂️
    APIs live in Git repositories, which means every tweak, enhancement, or late-night coding mishap is safely tracked. Rollbacks? Easy. History of bad decisions? Also easy.

  5. It’s Fun 🎉
    Let’s not forget the most important reason: building APIs is inherently satisfying. Designing schemas, structuring routers, seeing it all come together—it’s the developer equivalent of nailing a 3-pointer.

The Process of Extending the API

Here’s how I extended my FastAPI app to handle crypto coin data.

1. Create a New Feature Branch

I started with a feature branch. This isolates the changes so you don’t accidentally break your beautifully functioning main branch.

1

git checkout -b feature/crypto-coin

Now you’re in a pristine environment where you can unleash all your creativity—or chaos—without worrying about collateral damage.

2. Build the Crypto Domain

The new domain involved creating the following files:

Model: Defining the Tables

The database structure is the foundation of everything. For coins, coin_market_data, and coin_tags, I made the following design decisions:

  • Coins: A Type 1 Slowly Changing Dimension (SCD) table. Updates overwrite existing data, with timestamps (created_at and updated_at) tracking changes.
  • Market Data: A transaction table—new rows are added for every update. No overwrites here.
  • Tags: Simple many-to-one relationship with coins. Old tags are wiped and replaced during each update.

Here’s the SQLAlchemy model for coins:

1
2
3
4
5
6
7
8
9

class Coin(Base):
    __tablename__ = "coins"

    id = Column(Integer, primary_key=True)  # CoinMarketCap ID
    name = Column(String, nullable=False)
    symbol = Column(String, nullable=False)
    slug = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, nullable=False)

Repository: The Workhorse

This file encapsulates all database operations. Highlights include:

  • Upsert for coins: Handles both inserts and updates.
  • Batch insert for coin_market_data: Efficiently processes transaction data.
  • Replace operation for coin_tags: Deletes old tags and adds new ones in one go.

Schemas: Verbosity FTW

I went full throttle with Pydantic schemas, documenting every field with descriptions, examples, and constraints. Why? Because nothing says “I know what I’m doing” like detailed API docs that basically write themselves.

Example schema for CoinResponse:

1
2
3
4
5
6

class CoinResponse(BaseModel):
    id: int = Field(..., description="The unique ID of the coin.", example=1)
    name: str = Field(..., description="The name of the coin.", example="Bitcoin")
    symbol: str = Field(..., description="The symbol of the coin.", example="BTC")
    created_at: datetime = Field(..., description="Record creation timestamp.")
    updated_at: datetime = Field(..., description="Last update timestamp.")

Router: The API Layer

I built POST endpoints for:

  • Creating or updating coins
  • Inserting market data
  • Replacing tags

Each endpoint has clear summaries, detailed descriptions, and example payloads. For example, the replace_tags endpoint:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

@router.post(
    "/tags",
    response_model=None,
    summary="Replace tags for a cryptocurrency",
    description=(
        "Replace all tags for a cryptocurrency. Existing tags will be deleted "
        "and replaced with the new list provided in the request."
    )
)
async def replace_tags(coin_id: int, tags: list[CoinTagCreate], db: AsyncSession = Depends(get_db)):
    try:
        await replace_coin_tags(db, coin_id, [tag.tag for tag in tags])
        return {"message": "Tags successfully replaced."}
    except Exception as e:
        raise HTTPException(status_code=500, detail="Failed to replace tags.")

3. Add the Router to the App

With the domain built, the final step was adding the router to the app’s entry point in main.py:

1
2
3

from app.domains.crypto_coin.router import router as crypto_router

app.include_router(crypto_router, prefix="/crypto", tags=["Crypto Coins"])

This single line integrates the entire crypto domain into the app. Simple, clean, and satisfying.

What’s Next?

With the API layer done, it’s time to build an Airflow DAG to:

  1. Query the CoinMarketCap API for data.
  2. Send that data to the API using Python’s requests module.

Once that’s in place, I’ll have a complete data pipeline: from API to database. And after gathering enough data, the fun part begins: analysis 🚀.