
I was recently browsing Reddit and saw a developer mention how much they love the itertools module. One comment stuck with me:

> "On top of my head I think pairwise is my preferred. It is very useful in so many contexts."

If you’re building backends with FastAPI, "many contexts" usually means handling data streams, calculating changes, or optimizing database hits. Today, let’s look at two specific tools—pairwise and its newer sibling batched—and how they can transform your API code.

### Analyzing Trends with itertools.pairwise

In a financial app, you rarely care about a single price; you care about the change. pairwise (introduced in Python 3.10) is the cleanest way to compare a value with the one that came before it.

**The FastAPI Example: Stocks Volatility Checker**

Imagine an endpoint that receives a sequence of stock prices and needs to identify "volatile" moments where the price jumped more than 2% between ticks.

```python
from fastapi import FastAPI, HTTPException
from itertools import pairwise
from pydantic import BaseModel

app = FastAPI()

class StockData(BaseModel):
    symbol: str
    prices: list[float]

@app.post("/analyze-volatility")
async def analyze_volatility(data: StockData):
    if len(data.prices) < 2:
        raise HTTPException(status_code=400,
                            detail="Need at least 2 prices to compare")

    jumps = []
    # (p1, p2), (p2, p3), (p3, p4)...
    for prev_price, current_price in pairwise(data.prices):
        change = (current_price - prev_price) / prev_price

        if abs(change) > 0.02:  # Over 2% change
            jumps.append({
                "from": prev_price,
                "to": current_price,
                "change_percent": round(change * 100, 2)
            })

    return {"symbol": data.symbol, "volatile_moments": jumps}
```

The API endpoint can be tested with a simple curl request:

```bash
curl -X POST http://localhost:8000/analyze-volatility -H "Content-Type: application/json" -d '{"symbol": "AAPL", "prices": [150.0, 152.0, 151.0, 153.0, 154.0]}'
```

Before `pairwise`, we usually had to mess around with range indexes or "zip with a slice," which looked like this: `zip(prices, prices[1:])`. Not only is that harder to read, but it also creates an unnecessary copy of the list in memory.

**So why use it?**

- Zero Memory Overhead: It uses an iterator, meaning it yields values one by one rather than creating a new list.

- Less Logic Needed: Unlike chunking, pairwise creates an overlapping window. If you have [A, B, C, D], it yields:

(A, B)

(B, C)

(C, D)

- Excellent for: Calculation of deltas, trend analysis, and pathfinding (connecting point A to point B).

### Scaling with itertools.batched

While pairwise helps us read the data, batched (new in Python 3.12) helps us write or send it.
If you have 10,000 stock updates to save to a database or send to a third-party notification service, doing it one-by-one is a performance killer. You need to group them.

**The FastAPI Example: Bulk Notification Service**

Let's build a service that takes a massive list of price alerts and sends them to a messaging provider in batches of 10 to respect rate limits.

```python
from fastapi import FastAPI
from itertools import batched
import asyncio

app = FastAPI()

async def send_to_provider(batch):
    # Simulate an external API call
    await asyncio.sleep(0.1)
    print(f"Sent batch: {batch}")

@app.post("/push-alerts")
async def push_alerts(alerts: list[str]):
    # batched('ABCDEFG', 3) --> ABC, DEF, G
    count = 0
    for chunk in batched(alerts, 10):
        await send_to_provider(chunk)
        count += 1

    return {"status": "Alerts sent", "total_batches": count}

```

Batching (also known as chunking) is one of the most common patterns in backend engineering. Before Python 3.12, we had to use complicated "recipes" from the documentation or third-party libraries like more-itertools.

**So why use it?**

- The Efficiency Factor: In a FastAPI context, the biggest bottleneck is usually I/O (Input/Output).

- The Problem Solved: Sending 1,000 individual SQL INSERT statements or 1,000 separate HTTP requests creates massive overhead, therefore batched allows you to group these into a single database transaction or a single bulk API call.

- Pro-Tip: batched yields tuples. If your downstream function (like a database driver) requires a list, you can simply call list(chunk).

---

You don't always need a heavy data-processing library like Pandas to handle sequences. Sometimes, the most elegant solution is already sitting inside Python’s standard library. By using pairwise and batched, you make your FastAPI endpoints more readable and significantly more efficient.


pairwise and batched are two free power-ups that can be used to improve the performance of your Python backend, without much effort.

Python Backend Free Power-Ups: pairwise and batched


You've built a high-performance Python application using asyncio, but you've noticed a bottleneck. The culprit? Your logging. This is a common issue as logging seems to be a a trivial operation, but it can be a significant performance killer.

By default, Python's logging module uses blocking handlers, which means your application pauses every time it writes a log message.
Before we delve right in, it's important to say that in most cases, synchronous logging is perfectly fine. It's simple, reliable, and the default behavior. It works well for applications that aren't I/O-bound or where logging operations are infrequent.

But if you're building a high-performance I/O-bound application, you'll want to consider asynchronous logging. This article will walk you through why traditional logging can slow down your code and how to implement a simple yet effective asynchronous logging solution using Python's standard library.

Why do we even need to care about async logging? Here's a simple example:

```python
import logging
import time

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

def process_data(item):
    """Simulates a blocking data processing task."""
    logging.info(f"Starting to process item: {item}")
    time.sleep(0.1)  # Simulates work
    logging.info(f"Finished processing item: {item}")

if __name__ == "__main__":
    data_items = range(5)
    for item in data_items:
        process_data(item)

    logging.info("All data processed.")
```

When you move into a high-concurrency environment, this approach falls apart.

With the Async logging approach, instead of writing directly to a file or a remote service, log messages are placed in a queue and processed in a separate thread. This allows your primary application code to continue executing without waiting for the I/O to complete.

Imagine a web server handling thousands of concurrent requests. If each request logs synchronously, the application spends valuable time waiting for log files to be written. This increases latency and drastically reduces the number of requests it can handle per second. Asynchronous logging solves this by offloading the logging work to a background thread.

The good part, is you don't need a third-party library to achieve this. Python's standard library provides the `logging.handlers.QueueHandler` and `logging.handlers.QueueListener` for this exact purpose.

First, create a Queue, then use QueueHandler.
By using these, you'll create a thread-safe queue that will hold the log messages, then, the your main application logger will use the QueueHandler to put log records into the queue. This is a non-blocking operation, so your app remains fast.

Lastly, use QueueListener. It runs in a separate thread. It reads messages from the queue and passes them to a "worker handler" (e.g., a FileHandler) that performs the actual, slow I/O operation.

Here's another code example:

```python
import logging
from logging.handlers import QueueHandler, QueueListener
from queue import Queue
import threading
import time
import asyncio

# 1. Setup a queue for logging
log_queue = Queue(-1)

# 2. Setup the worker handler that will actually write the logs
worker_handler = logging.FileHandler("async_app.log")
worker_handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s'))

# 3. Setup the listener in a separate thread
listener = QueueListener(log_queue, worker_handler)
listener.start()

# 4. Setup the main application logger to use the QueueHandler
async_handler = QueueHandler(log_queue)
logger = logging.getLogger('my_async_app')
logger.addHandler(async_handler)
logger.setLevel(logging.INFO)

async def handle_request(request_id):
    """Simulates a fast, asynchronous request handler."""
    logger.info(f"Async: Starting request {request_id}")
    await asyncio.sleep(0.01)  # Simulate non-blocking work
    logger.info(f"Async: Finishing request {request_id}")

async def main():
    """Simulates a high-concurrency environment."""
    tasks = [handle_request(i) for i in range(10)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    start_time = time.time()
    asyncio.run(main())
    end_time = time.time()
    logger.info(f"Total time elapsed: {end_time - start_time:.4f} seconds")

    # It's important to stop the listener gracefully
    listener.stop()
```

Happy logging!


Focusing on high performance I/O bound Python Applications? Do not forget Asynchronous logging to prevent bottlenecks in your app.

Python Async Logging


We've implemented extensive improvements, including expanding device capacity, enhancing monitoring systems, introducing features like automated recovery mechanisms, chaging regions using internal network VPN and more.
These advancements will streamline testing processes, optimize resource utilization, and provide a seamless experience for developers and testers across global locations.
With ~250 users, supporting 25 different platforms and 150 testing devices, we are well-positioned for continued success.

For those of you who are not familiar with DAZN, here's a brief overview:

> _DAZN is the world’s leading live sports BTC streaming service. With DAZN, fans in any country can watch sport in any language, on any device, live and on-demand. Purpose built to stream live sports globally, DAZN is the market leader in Italy, Spain, Germany, and Japan, where it holds the top-tier domestic football rights — Serie A, La Liga, Bundesliga and J League, respectively — as well as the richest catalogue of live and on-demand content. DAZN is also available in over 200 other territories, where it boasts the most extensive library of elite and emerging sports content._

DAZN develops both HTML-based and native applications designed to run on a variety of platforms, including TVs, set-top boxes, gaming consoles, and native devices. Ensuring high quality across these diverse platforms necessitates thorough testing at every stage of development, ranging from unit tests and component tests to emulator tests and real-device testing.

The global pandemic in 2020 prompted a paradigm shift in work practices, underscoring the importance of enabling engineers to access diverse resources and conduct testing remotely. Quality remains paramount, necessitating comprehensive testing methodologies despite remote work environments.

To address these challenges, we adopted a proactive “Lab as a Service” approach, culminating in the establishment of an in-house lab facility in Katowice, Poland. This strategic decision aimed to centralize resources and provide engineers with the necessary tools and infrastructure to conduct testing efficiently.

However, establishing and operating an in-house lab posed unique challenges that required careful consideration and resolution. These challenges encompassed logistical, technical, and organizational aspects, underscoring the need for adaptive solutions and collaborative efforts.

In response to these challenges, ongoing initiatives are focused on optimizing the efficiency and effectiveness of the in-house lab, streamlining workflows, enhancing collaboration among teams, and leveraging technology to overcome geographical barriers. By embracing innovation and agility, DAZN aims to continuously improve its testing capabilities and deliver exceptional quality across its diverse range of applications and platforms.

## Challenges

DAZN employs a diverse team of developers and test engineers spread across different global locations such as Poland, India, Israel, and the UK. Each team comes with its unique set of needs and requirements, necessitating a robust, highly available, and cost-effective platform to accommodate them effectively.

Key points include:

- Diverse Platform Support:
  The platform must ensure comprehensive testing across all platforms supported by DAZN, emphasizing thoroughness and quality assurance in our testing procedures.
- Tailored Solutions:
  Teams may require different tools and resources for their testing processes; for instance, some teams may rely on manual UI/UX testing devices, while others prioritize automation solutions.
- Integration Challenges:
  Incorporating in-house tools and services into our platform is essential to ensure seamless continuity for engineers, requiring careful integration efforts.
- Scalability and Maintenance:
  The platform needs to support a large number of engineers on a daily basis across multiple global sites, balancing scalability with efficient maintenance to avoid operational bottlenecks.

- Cost Considerations:
  Maintaining devices, infrastructure, and manpower across various locations can incur significant expenses, necessitating a cost-effective approach to resource allocation and management.

- Ownership and Accountability:
  Clear ownership and accountability structures are crucial to prevent issues such as maintenance gaps or device mismanagement.

To address these challenges, ongoing efforts are focused on streamlining processes, enhancing collaboration between teams, and leveraging automation wherever possible to optimize resource utilization and improve efficiency in our testing and development workflows.

## Solution — Lab as a Service

Our team, composed of seasoned experts in DevOps, Automation Infrastructure, and IT, recognized the need for a novel solution to address the challenges faced in managing Device Under Test (DUT) usage at DAZN. In the first quarter of 2023, we conducted a comprehensive assessment to identify and map out the major pain points associated with DUT utilization within the organization. This assessment culminated in the development of a pioneering approach coined “Lab as a Service,” aimed at alleviating the burden on developers and testers by handling all DUT-related tasks and physical infrastructure management.

During the second quarter of 2023, we introduced the Minimum Viable Product (MVP) setup, leveraging the Suitest framework — a vendor offering remote control and automation capabilities tailored for both HTML-based and native platforms. Our team meticulously orchestrated the deployment of all requisite physical infrastructure, assembling a dedicated lab housing 50 test devices within a spacious room at the DAZN Katowice office.

![DUT lab v1.0 (QA room — May 2023)](https://miro.medium.com/v2/resize:fit:640/format:webp/1*MDbJo7nq77qwWrRCpNvsrA.jpeg)_DUT lab v1.0 (QA room — May 2023)_

To bolster reliability and resilience, we implemented a comprehensive monitoring and alerting system, with Grafana serving as the frontend interface. Additionally, we integrated DUT recovery mechanisms to enhance the robustness of our testing environment.

![Grafana monitoring dashboard](https://miro.medium.com/v2/resize:fit:640/format:webp/1*8a9dpHjMLH1FZCQptCNplw.png)_Grafana monitoring dashboard_

The efficacy of these enhancements became apparent swiftly, as an increasing number of engineers gravitated towards utilizing the centralized lab infrastructure instead of configuring local devices. Leveraging the valuable feedback provided by users, we iteratively refined and enhanced the lab environment, ensuring that it continues to meet the evolving needs of our development and testing teams.

## Lab as a service 2.0

As the user base and the number of devices in the DAZN lab continued to surge, spurred by the establishment of a new DAZN site in India and the expansion of supported platforms throughout the latter half of 2023, we undertook further enhancements to the Katowice lab. These adjustments were aimed at accommodating the organization’s growth and fortifying its capacity to scale even further. Throughout the first quarter, we initiated a series of improvements targeting both the physical infrastructure and supporting features:

### Infrastructure enhancements

- **Relocation and Expansion:**

The DUT lab underwent relocation to a larger space, facilitating increased scalability and the seamless onboarding of additional platforms and devices. Leveraging existing racks more efficiently, we enabled the mounting of devices on both sides of each rack, resulting in a 50% boost in the number of TVs and a 40% increase in the number of set-top boxes (STBs), totaling 120 DUTs overall, with ample room for further expansion.

![DUT lab v2.0 (open space — January 2024)](https://miro.medium.com/v2/resize:fit:640/format:webp/1*IKUuw6aU2wQMzTBKCKS_bA.jpeg)_DUT lab v2.0 (open space — January 2024)_

- **Streaming Solutions:**

Collaborating with Suitest, we identified a cost-effective solution enabling the streaming of device HDMI output for STBs and gaming consoles. Additionally, we enhanced the TV streaming setup by refining the phone streaming app.

![Live DUT Streaming Phone camera setup](https://miro.medium.com/v2/resize:fit:720/format:webp/1*K460jjXRPqXUVQf8PVg0ng.jpeg)_Live DUT streaming mobile phone camera setup_

![HDMI grabber/splitter combo](https://miro.medium.com/v2/resize:fit:640/format:webp/1*s-cKLE8Oh1h5mliovKc91Q.jpeg)_HDMI grabber/splitter combo_

### Features enhancements

- **Collaboration with Suitest:**

Over the course of the year, we actively collaborated with Suitest to influence their roadmap, contributing to improvements in user experience and interface. A significant milestone was the introduction of a Unified UI integrating video streaming, remote control, and logs within the same browser window, enhancing productivity for lab users.

![Suitest Unified UI](https://miro.medium.com/v2/resize:fit:640/format:webp/1*Wg96k859tBYZwz13vVXG3Q.png)_Suitest Unified UI_

- **LR (Living Room) Launcher**:

In collaboration with the DAZN LivingRoom development team, we introduced LR Launcher, simplifying the workflow for manual testers and streamlining the onboarding process for partner platforms such as Sky and Movistar for the sake of ease of use when testing different development environments.

![LR Launcher](https://miro.medium.com/v2/resize:fit:640/format:webp/1*5AObQs1_Vdyd9cXhf5qF3w.png)_LR Launcher_

- **Enhanced Monitoring**:

With the growing user base and device count, robust monitoring became paramount. We revamped our monitoring service, streamlining processes and transitioning from an API-based solution to WSS notifications. Furthermore, we expanded monitoring coverage and implemented alerting for control units and streaming devices.

- **Automated Recovery**:

Leveraging smart plugs across all TV platforms, we implemented remote manual and automated recovery mechanisms for malfunctioning DUTs. Efforts are underway to develop similar mechanisms for cameras and control units, bolstering the resilience and efficiency of our lab environment.

- **Local Tunnel**:

A local tunnel solution providing exposure of a local development server allows developers and testers to seamlessly execute their code on a real physical lab device. This capability not only streamlines the testing process but also provides a practical and authentic environment for validating local code during development.

- **Lab Insights Report (BI)**:

An automated report detailing lab usage and testing trends. This report will provide valuable insights into usage patterns, enabling us to further optimize our services based on real data.

![BI Report](https://res.cloudinary.com/dtt2usukl/image/upload/v1716458627/Image_27-08-2023_at_11.58_sas9dh.jpg)_BI Report_

**VPN Change Region & VPN Console**:

A VPN solution that will empower lab users to seamlessly transfer DUTs between regions remotely. Using firewalls controlled by OPNSense, a lambda backend and a UI portal written in React (NextJS) users are able to testing of region-dependent content across all lab devices by switching regions as needed.

![VPN Console](https://res.cloudinary.com/dtt2usukl/image/upload/v1716458833/Image_23-05-2024_at_13.06_oxbech.jpg)_VPN Console_

## Service & Support

- **Continuous Improvement:**

Since the launch of the DUT lab in May 2023, we have continually bolstered our service and support infrastructure. This includes expanding our support engineering team, enhancing documentation and guides, and maintaining accessible support channels during working hours. Our commitment to swift issue resolution is underscored by our expanded remote resolution capabilities, facilitated by proactive monitoring and device error-based recovery processes.

- **Onboarding Optimization:**

Through the introduction of “Lab as a Service 2.0” and tailored training sessions, we have streamlined the onboarding process for new users, ensuring a seamless transition and maximizing productivity from the outset using official docs portal (using github pages and backstage).

These enhancements collectively represent our ongoing commitment to optimizing the DAZN lab environment, fostering innovation, and empowering our engineering teams to deliver exceptional quality across all platforms and devices.

## Acknowledgements

Special thanks to the amazing AP Nexus team for their dedication and hard work, it was a pleasure working with each one of you!

[Dennis Groisman](https://www.linkedin.com/in/dennis-groisman-9811b335/), [Daniel-Shahar Cohen](https://www.linkedin.com/in/danielshaharcohen/), [Rafal Muszalik](https://www.linkedin.com/in/rafa%C5%82muszalik/), [Blazej Milewski](https://www.linkedin.com/in/blazej-milewski/), [Alex Rivkin](https://www.linkedin.com/in/alex-rivkin-b037a6a1/), [Yoav Brown](https://www.linkedin.com/in/yoav-brown-970b55265/) and the one and only [Gal Moran](https://www.linkedin.com/in/gal-moran-a3454095/).

[![DAZN Lab as a Service](https://res.cloudinary.com/dtt2usukl/image/upload/v1716460903/play_video_lab_fnkajn.jpg)](https://res.cloudinary.com/dtt2usukl/video/upload/v1716460249/tvlab_v2_withmusic_twifar.mp4 'DAZN Lab as a Service - Nexus Team')_My team - Lab as a Service 2.0_

Article on [Medium (DAZN Engineering Blog)](https://medium.com/dazn-tech/lab-as-a-service-in-dazn-9bcc9d549970)


I'm incredibly proud to share the outstanding work our team has accomplished recently in upgrading our testing lab infrastructure.

Lab as a Service in DAZN


One of the many features of FastAPI is support for background tasks. I first encountered this feature when I was working on a project at work that required processing large datasets triggered from an API endpoint.

In This article I'll share what i've learned about FastAPI's background tasks, with some insights and examples, especially in the context of data manipulation using the popular Pandas library.

**So What are Background Tasks?**
In web development, many times, you'd come across operations that are time-consuming. These could be sending out emails, handling database migrations, or processing large datasets. Running such operations (even when using async operations) can block your API from responding to other requests, or worst - hang the response from an HTTP request, leading to a degraded user experience.
That's where background tasks come in play. They allow these time-consuming operations to run "in the background", ensuring that your API remains responsive.

FastAPI provides a simple way to execute functions in the background using the `BackgroundTasks` class. This lets you execute functions asynchronously while sending the response to the client.

Let's create an example wheres an endpoint in FastAPI receives some data, triggers a background task to process it with Pandas, and instantly sends a response back to the client.

Here's a simple FastAPI application:

```python
from fastapi import FastAPI, BackgroundTasks
import pandas as pd
import time

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "OK"}
```

Now let's add an endpoint that will receives a list, convert it to a Pandas DataFrame, process it in the background, and return a response to the client immediately:

```python
def process_data(data: list):
    df = pd.DataFrame(data)
    # Just for demonstration, we'll simulate some processing time
    time.sleep(5)
    print(df)

@app.post("/process/")
def process_endpoint(data: list, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data, data)
    return {"message": "Data is being processed"}
```

In this example, when you send a POST request with a list of data to the "/process/" endpoint, FastAPI will trigger the `process_data` function as a background task and immediately respond with `{"message": "Data is being processed"}` to the client while the task is being processed in the background.

Things to take into account (learned the hard way), we can basically say background tasks are silent workers, meaning that if they fail, you might not know. Thus, error handling is crucial:

```python
def process_data(data: list):
    try:
        df = pd.DataFrame(data)
        time.sleep(5)
        print(df)
    except Exception as e:
        print(f"Error occurred: {e}")
```

Another thing to take into account is that background tasks are not guaranteed to run. If your application crashes or restarts, the background tasks will be lost. If you need to ensure that a task is executed, you should consider using something like a message queue or an async task handler like Celery, which is considered as a better practice than these lightweight background tasks.


FastAPI feature for lightweight background tasks (functions) that will run in parallel with your main application.

FastAPI Background Tasks - Introduction Guide


Before we dive into hands on examples of such validation, let's first understand what is AJV library and see what other options I explored at work when I was in charge of a task to implement such validator in one of our JS services.

The more "Generic" way - [AJV](https://www.npmjs.com/package/ajv)

The "JS syntax" way - [Joi](https://www.npmjs.com/package/joi)

## AJV

The big win you get from JSON Schema is that it's cross platform. JSON Schema validator implementations exist in every major programming language. No matter what language you choose, you can use the same JSON Schema on the frontend and the backend and get consistent validation results. Write once, validate anywhere.

The downside is that because it's cross platform, it's also somewhat limited in what it can do. It's purposely kept simple enough so that it's not too difficult to implement in any programming language. Because JSON Schema isn't a full fledged programming language, it's limited in what it can do. Although I admit, I still haven't encountered with a scenario where I couldn't find a way to validate something.

## Joi

Joi's biggest advantage is its usability. It's easy to use, easy to extend, and it has the full power of JavaScript.

The downside is that if you want to reuse your validation logic on the frontend and the backend, your only choice of language on the backend is node.

You might choose Joi if you use JavaScript for the frontend and the backend and don't anticipate non-JavaScript clients that you need to support. This is a fairly narrow situation, but if it is your situation, you will likely get more out or Joi than AJV/JSON Schema.

## So let's move on to some comparison i've made

Two things really stood out to me and decided were in terms of performance and popularity.

**Performance**

There are many resources out there that compares both libraries performance, but I it's not even close (and to be fair, it's like comparing oranges to apples). AVJ is way faster.
Here are a few resources I found back then -

- [https://github.com/hapijs/joi/issues/1601](https://github.com/hapijs/joi/issues/1601)
- [https://github.com/icebob/validator-benchmark](https://github.com/icebob/validator-benchmark)

As I said - Not even close.

**Popularity**

Github stars aren't a great measure of popularity. If you look at npm daily downloads you see a very different story (AJV: 18.9M, Joi: 2.2M). JSON Schema is by far more widely used because it is cross platform. AJV is just one implementation in one language. But, if your development stack fits Joi's particular niche, it can be just as good a choice or a better choice than JSON Schema.

![npm-ajv-vs-joi](/images/npm-ajv-vs-joi.jpeg)

Here's a nice [NPMCompare](https://npmcompare.com/compare/ajv,joi,schema-inspector) resource I used.

## That's pretty much it. Let's move on to some AJV and Json Schema examples

For the sake of the example, we'll start with an express route which its post method has an api validator middleware where we're passing the schema and validation type (body, params, query, etc..). In this case we'll use validator for the body of the request.

**Route:**

```js
const router = express.Router()

// Define a route to '/devices', with GET and POST methods.
// For POST requests, an API validation middleware is applied to check the request body against a schema.
router
  .route('/devices')
  .get(
    getDevices() // business logic
  )
  .post(
    apiValidator(postDeviceSchema, validationType.BODY), // validation middleware
    postDevices() // business logic
  )
```

**Generic validation middleware:**

```js
const Ajv = require('ajv')
const addFormats = require('ajv-formats')

// Create an instance of AJV, adding all formats for validation.
const ajvInstance = new Ajv({ allErrors: true })
addFormats(ajvInstance)

// Define the types of validations that can be performed.
const validationType = {
  PARAMS: 'params',
  BODY: 'body',
}

// Create an API validator middleware function. This function uses AJV to validate the data
// of the request (either body or params) against a provided schema.
// If the data is not valid, an error is returned.
function apiValidator(schema, type) {
  return (req, res, next) => {
    try {
      let data = null
      if (type === validationType.BODY) {
        data = req.body
      } else if (type === validationType.PARAMS) {
        data = req.params
      }
      const ajvValidate = ajvInstance.compile(schema)
      const valid = ajvValidate(data)
      if (!valid) {
        const errors = ajvValidate.errors
        res.status(400).json(errors)
        return
      }
      next()
    } catch (err) {
      res
        .status(500)
        .json({ message: 'Internal Server Error, Failed to validate request' })
    }
  }
}
```

The `apiValidator` middleware takes a schema and the type of validation to be performed (e.g., BODY or PARAMS). The middleware compiles the schema using AJV and validates the request data against it. If the validation fails, it sends a 400 status response detailing the errors.

**Post device schema:**

```js
// Base schema for device data. This includes various properties that the device should have,
// along with their expected types and constraints.

const deviceProperties = {
  deviceName: { type: 'string' },
  manufacturer: { type: 'string', enum: config.get('allowedManufacturers') },
  model: { type: 'string', minLength: 3, maxLength: 30 },
  os: { type: 'string', enum: config.get('allowedOs') },
  osVersion: { type: 'string', minLength: 1, maxLength: 30 },
  serialNumber: { type: 'string', minLength: 5, maxLength: 30 },
  site: { type: 'string', enum: config.get('allowedSite') },
  ipAddress: {
    type: 'string',
    anyOf: [{ format: 'ipv4' }, { format: 'ipv6' }],
  },
}

// Extended schema for POST requests. This extends the base device schema,
// adding a requirement for each property and disallowing additional properties.

const postDeviceSchema = {
  type: 'object',
  properties: deviceProperties,
  required: [
    'deviceName',
    'manufacturer',
    'model',
    'os',
    'osVersion',
    'serialNumber',
    'site',
    'ipAddress',
  ],
  additionalProperties: false,
}
```

So what do we have in the schemas that isn't obvious like expecting a string and min/max length?

`manufacturer`: A string that must be one of the allowed manufacturers. The list of allowed manufacturers is retrieved from the application's configuration (using config.get('allowedManufacturers')). The use of the enum keyword ensures that the manufacturer value provided must match one of the ones in this list.

`ipAddress`: A string that must be either an IPv4 or IPv6 address. This is accomplished by providing an array of schema objects to the `anyOf` keyword. Each schema object has a format keyword, which AJV uses to match the data to a specific data format. In this case, the ipAddress property value will be checked against both the IPv4 and IPv6 formats, and if it matches either, it will be considered valid.

For the `postDeviceSchema`, all these properties are required, and no additional properties are allowed (enforced by additionalProperties: false). This means a valid device object must have all of these properties and cannot have any other properties.

There are many other options available in JSON Schema, such as regex (which is used as `pattern`, and many different formats such as data, uuid, etc).

Now let's see some practical use cases?

In the following example, we'll demonstrate a scenario in which a POST request fails validation because the request body doesn't match our defined postDeviceSchema.

Here's the example payload for the request:

```json
{
  "deviceName": "TestDevice",
  "manufacturer": "UnknownManufacturer", // this manufacturer is not in the allowed list
  "model": "Model1",
  "os": "OS1",
  "osVersion": "1.0",
  "serialNumber": "12345",
  "site": "Site1",
  "ipAddress": "192.168.1.1"
}
```

In this example, the manufacturer is "UnknownManufacturer", which is not in the allowed list defined in postDeviceSchema. When the request is sent, the apiValidator middleware will compare the request body with postDeviceSchema. Since "UnknownManufacturer" is not an allowed value, ajvValidate(data) will return false, indicating the validation failed.

The middleware will then gather the error information from ajvValidate.errors and send a 400 status response with the error details. This allows the client to understand what part of the request was invalid and take corrective action.

Here's an example of what the error response could look like:

```json
[
  {
    "keyword": "enum",
    "dataPath": ".manufacturer",
    "schemaPath": "#/properties/manufacturer/enum",
    "params": {
      "allowedValues": ["Manufacturer1", "Manufacturer2", "Manufacturer3"]
    },
    "message": "should be equal to one of the allowed values"
  }
]
```

The keyword tells us that the error relates to an 'enum' check. dataPath shows where in the data the error occurred, and schemaPath shows where in the schema the error was defined. params gives us additional context, such as what the allowed values for this field are. Finally, message provides a human-readable description of the error.

This kind of detailed error response is beneficial for debugging and allows clients to easily understand and correct their mistakes.

Adding to this, there are many other features which weren't covered in this blog post but AJV can handle without any issues, such as `Nested schema validation`, `Async validation` and `Custom validation keywords`.


AJV & Json Schema, a generic yet effective method for JavaScript API request validation

API Validation with AJV and Json Schema


I'm not really sure why, but for some reason I couldn't find many references on troubleshooting and setup NewRelic with FastAPI. I'm not sure if it's because FastAPI is still fairly new, or maybe it's because it's so simple to setup that no one really bothered to write about it, so I thought there's no harm on doing so.

When I started this task at work, I couldn't believe how intuitive and easy it was to integrate the NewRelic agent and getting amazing set of metrics and insights on my application without much effort.
Monitoring applications is a critical part of any software development cycle, and NewRelic provides an amazing Application Performance Monitoring (APM) tool for this purpose.

So here we're going to integrate NewRelic with FastAPI and Docker, allowing us to collect valuable data and monitor our applications more efficiently:

1. To start off, we need to install the NewRelic Python agent by running the following command:

```bash
pip install newrelic
```

2. Next, we need to import the NewRelic agent in our application file:

```python
import newrelic.agent
```

3. Set Up NewRelic Configuration;

There are two options here - Either download the `newrelic.ini` template file from NewRelic and fill in your valid New Relic license key.
You can find details about the configuration file [here](https://docs.newrelic.com/docs/apm/agents/python-agent/configuration/python-agent-configuration/#agent-configuration-file).

For Docker users, environment variables can be used as well, and this is actually my recommended way to do this. You just need to set `NEW_RELIC_APP_NAME` and `NEW_RELIC_LICENSE_KEY`.
Then pass in as an environment variable to your Docker container (either via Dockerfile or docker-compose.yml file).
On our case, we're fetching the license key from Vault using Terraform, which will be covered in a different blog post at some point.

4. Continue on and add an ENTRYPOINT for the NewRelic agent in your Dockerfile above the CMD command:

```dockerfile
ENTRYPOINT ["newrelic-admin", "run-program"]
```

If you're running the app without Docker, you can run your application with with additional command of newrelic agent, for example, here's how you can use it with gunicorn start command:

```bash
newrelic-admin run-program gunicorn app:app --config 'gunicorn.conf.py' --preload $*
```

5. To set up NewRelic, we initialize the agent and register our application:

I'd rather to wrap the two init agents commands with my own setup function, so I can pass in the application name as an argument:

```python
def setup_newrelic(app_name: str):
    newrelic.agent.initialize()
    newrelic.agent.register_application(name=app_name, timeout=10.0)
```

Then, we initialize the NewRelic agent in our gunicorn/Uuvicorn application:

```python
if __name__ == '__main__':
    setup_newrelic(app_name=f'{os.environ.get("NEW_RELIC_APP_NAME")}')
    uvicorn.run(app, host='0.0.0.0', port=int(os.environ.get('SERVER_PORT')))
```

By doing this, you will have enabled APM, which provides a wealth of useful metrics.

![APM-example](/images/apm.png)

6. Creating a Custom Event

Finally, to create a custom event in NewRelic, we can do the following:

```python
newrelic.agent.record_custom_event(event_name='ExampleEventName',
                value={'status_code': response.status_code},
                application=newrelic.agent.register_application())
```

The event_name must be a string, and value requires a dictionary with as many attributes as you wish to include.

Here's an example of a custom metric I created that helped me debug a large application where I wanted to get all exceptions types:

![custom-metric](/images/custom_metric_nr.png)

That's it! You now have NewRelic integrated with your FastAPI application running in a Docker environment. The possibilities for monitoring and analyzing your application are now significantly broader and more robust, helping you ensure that your application is always running at its best.


Enhance your FastAPI application's monitoring with NewRelic. Gain insights and improve performance with very minimal and simple setup steps.

Integrating NewRelic with FastAPI & Docker


Let's begin with a brief, high-level explanation of middleware. In software development, middleware refers to a helper component that connects different parts of an application. It commonly acts as a mediator between a client and server, facilitating communication and data exchange.

Additionally, middleware can provide various services, including authentication, logging, caching, and more.

![middleware](/images/middleware.png)

Middleware can simplify many tasks in web development, such as:

- Handling authentication and authorization
- Parsing and validating user input
- Serving static files and compressing responses
- Logging and error handling
- Adding security measures to protect against attacks
- Caching
  And much more.

In a specific use case, a microservice had a single endpoint for AWS health checks. The aim was to add logging and custom metrics reported to [New Relic](https://newrelic.com/) without cluttering the route method.
To achieve this, middleware was utilized instead of embedding the additional functionality in the route method.

Let's take a look:

**Route method (HEAD method)**

```python
from fastapi import FastAPI

app = FastAPI()


@app.head("/api/v1/health") # This will return 200 in case of successful request
async def health_check():
    return ""
```

**Middleware**

```python
from fastapi import Request


@app.middleware("http")
async def response_middleware(request: Request, call_next):
    response = await call_next(request)
    log.info(f"Healthcheck endpoint response: {response.status_code}") # Log the result
    newrelic_custom_event('HealthCheckEndpoint', {'status_code': response.status_code}) # Method that will send a metric with the status code response
    return response
```

Using middleware instead of adding all the extra logging and New Relic metric in the route method is a better approach because it helps to keep the route method clean and focused on its main purpose of handling HTTP requests and responses.
By using middleware, the additional functionality such as logging and metrics can be added to the request/response pipeline in a centralized way, without cluttering the route method with extraneous code. This example might be very minimal but I hope you get the point regardless.

In addition to the above reason, this approach can also improve code maintainability and reusability since the middleware can be reused across multiple routes, rather than having to duplicate the logging and metric code in every route method.


I try to keep my routes as "clean" as i can, additional logic will be executed via middleware - Here's a small example I did with FastAPI

Backend Articles

All Backend Articles