In this article, we will create a FastAPI application. It retrieves real-time football match data from the Football Data API. The application stores the data in an Elasticsearch database. By the end of this tutorial, we will have a functional API that allows us to retrieve and search for football matches.

Understanding FastAPI

FastAPI is a modern, high-performance web framework for building APIs with Python 3.6+, based on standard Python type hints. Here are some key features:

• Asynchronous Support: FastAPI utilizes asynchronous programming, enabling high concurrency and minimal latency, allowing your API to handle numerous requests simultaneously without blocking.
• Automatic Documentation: It automatically generates OpenAPI and JSON Schema documentation, making it easy for developers to understand how to interact with your API.
• Type Checking and Validation: By utilizing Python-type hints, FastAPI validates incoming data and provides clear error messages if the data does not match the expected format.
• Dependency Injection: FastAPI supports dependency injection, allowing you to manage dependencies like database connections easily.

Elasticsearch Integration

Elasticsearch is a distributed, RESTful search and analytics engine. It is capable of quickly storing and indexing large volumes of data in real-time. Integrating FastAPI with Elasticsearch enables you to develop robust applications for efficient data storage, search, and analysis.

How FastAPI Works with Elasticsearch

1. HTTP Requests: FastAPI defines endpoints (like /ingest-live-football-data and /search-matches) that handle HTTP requests. When a request is made to these endpoints, FastAPI executes the associated function.
2. Data Fetching and Ingestion:

• In our application, we fetch football data using the requests library.
• The fetched data is then indexed in Elasticsearch using the es.index() method, allowing it to be searchable.

3. Searching Data:

• The /search-matches endpoint demonstrates how to query Elasticsearch for specific matches using the es.search() method.
• FastAPI handles the incoming query, interacts with Elasticsearch, and returns the search results to the client.

http://www.football-data.org API Key Generation

1. Go to Football Data API.
2. Sign up for an account.
3. Navigate to your account dashboard to find your API key.

Step 1: Setting Up Your Environment

Create a new directory for your project and navigate into it:

mkdir football-data-ingestion
cd football-data-ingestion

Step 2: Creating the FastAPI Application

1. Install FastAPI and Elasticsearch: Create a requirements.txt file with the following content:

fastapi
uvicorn
elasticsearch
requests

Then install the dependencies using:

pip install -r requirements.txt

2. Create the Application Structure:

football-data-ingestion/
├── app/
│   ├── main.py
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

3. Implement the FastAPI Code: Open main.py and add the following code:

from fastapi import FastAPI
from elasticsearch import Elasticsearch
import requests
import os

app = FastAPI()

# Elasticsearch Cloud Connection using environment variables
es = Elasticsearch(
    hosts=[os.getenv("ELASTICSEARCH_HOST")],
    http_auth=(os.getenv("ELASTICSEARCH_USERNAME"), os.getenv("ELASTICSEARCH_PASSWORD"))
)

FOOTBALL_API_URL = "https://api.football-data.org/v2/matches"
API_KEY = os.getenv("FOOTBALL_API_KEY")

# Root route
@app.get("/")
def read_root():
    return {"message": "Welcome to the Football Data Ingestion API"}

# Ingest real-time football data into Elasticsearch
@app.get("/ingest-live-football-data")
def ingest_football_data():
    headers = {"X-Auth-Token": API_KEY}
    response = requests.get(FOOTBALL_API_URL, headers=headers)
    
    if response.status_code != 200:
        return {"error": "Failed to fetch data from API"}

    match_data = response.json()

    # Ingest data into Elasticsearch
    for match in match_data['matches']:
        es.index(index="football-data", document=match)

    return {"status": "Data ingested"}

# Search API for football matches
@app.get("/search-matches")
def search_matches(team_name: str):
    query = {
        "query": {
            "match": {"homeTeam.name": team_name}
        }
    }
    result = es.search(index="football-data", body=query)
    return {"matches": result['hits']['hits']}

Code Explanation

• Elasticsearch Connection: The app connects to Elasticsearch using credentials stored in environment variables, making it secure and easy to configure.
• Data Fetching: The /ingest-live-football-data endpoint fetches match data from the Football API using the provided API key.
• Data Storage: Matches are indexed into Elasticsearch for easy searching, allowing users to query based on team names via the /search-matches endpoint.

Step 3: Creating Docker Configuration

1. Create a Dockerfile: Add the following content to your Dockerfile:

FROM python:3.9-slim

WORKDIR /app

COPY . /app

RUN pip install --no-cache-dir -r requirements.txt

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

2. Create a Docker Compose File: Create docker-compose.yml:

version: '3.8'

services:
  fastapi-app:
    build: .
    ports:
      - "8000:80"
    environment:
      ELASTICSEARCH_HOST: "Hostname"
      ELASTICSEARCH_USERNAME: "elastic"
      ELASTICSEARCH_PASSWORD: "your_password_here" # Replace with your actual password
      FOOTBALL_API_KEY: "your_football_api_key_here" # Replace with your actual API key

Step 4: Build and run the application

docker-compose up --build -d

Step 5: Validating Data Ingestion

After ingesting data, you can verify if the data is stored in Elasticsearch:

• Search for Matches: Use the /search-matches endpoint by sending a GET request, e.g., http://localhost:8000/search-matches?team_name=Manchester City.

Conclusion

In this article, we have developed a FastAPI application that retrieves and stores football data in Elasticsearch. This project demonstrates the capabilities of contemporary web frameworks and databases, laying a strong groundwork for future improvements like user authentication and data visualization.

The project code is available on my GitHub.

Related Article:

OpenTelemetry with Elastic Observability

Elastic RUM (Real User Monitoring) with Open Telemetry (OTel).

OpenTelemetry: Automatic vs. Manual Instrumentation — Which One Should You Use?

Configuration of the Elastic Distribution of OpenTelemetry Collector (EDOT)

Instrumenting a Java application with OpenTelemetry for distributed tracing and integrating with Elastic Observability

Test and Analyze OpenTelemetry Collector processing

You can also read this article on Medium.