I ran into this firsthand while trying to build a live options dashboard. HTTP requests weren't going to cut it, and everything I was reading seemed overly complex until I went back to the basics. This article is the result of that process.

We'll cover Python's websockets library from scratch, then move into FastAPI, where many Python backends live. It's worth noting that WebSockets aren't the only solution for real-time communication. WebRTC may be a better fit depending on your use case, but understanding WebSockets is the right starting point before exploring further.

Table of Contents

1. WebSocket Connections and Methods

2. How to Build Your First WebSocket in Python

3. File Transfer Over WebSockets

4. How to Connect to an External WebSocket

5. WebSockets in FastAPI

6. How to Handle WebSocket Disconnections in FastAPI

7. Conclusion

WebSocket Connections and Methods

A WebSocket connection enables bi-directional communication between a client and a server. Once a connection is established, both sides can communicate freely without either having to ask first. This is different from a regular HTTP request, where the client always has to ask before the server can respond.

It looks something like this:

        CLIENT  <===== open connection =====>  SERVER

Note that a WebSocket URL is not a regular web page, so you can't "visit it" like a website. You need a client to talk to it.

Different frameworks provide different methods for handling WebSocket connections. With Python’s websockets library, for instance, a connection is automatically accepted the moment a client connects. With frameworks like FastAPI, you have to explicitly call await websocket.accept(), otherwise the connection gets rejected.

Let’s look at the core methods provided by Python’s websockets library:

1. websockets.serve(...):  starts a WebSocket server.

2. websockets.connect(...): connects to a WebSocket server.

3. websockets.send(...): sends a message from either side.

4. websockets.recv(): receives a message from client or server.

recv() takes no arguments because it's purely a waiting operation. It waits for the next message and returns it:

message = await websocket.recv()

How to Build Your First WebSocket in Python

Before we dive into frameworks, let’s explore Python’s websockets library. You’ll set up a simple server and client, and exchange messages over a WebSocket connection, giving you a solid foundation for understanding WebSockets under the hood.

Environment Setup

Run the following in your virtual environment to install or verify the WebSockets package:

pip install websockets
# or, to check if it's already installed:
pip show websockets

Create the WebSocket Server

Create server.py in your project folder, and paste this:

import asyncio
import websockets

async def handler(connection):
    print("Client connected")

    message = await connection.recv()
    print("Received from client:", message)
    await connection.send("Hello client!")


async def main():
    async with websockets.serve(handler, "localhost", 8000):
        print("Server running at ws://localhost:8000")
        #await asyncio.Future()  # runs forever
        await asyncio.sleep(30)

asyncio.run(main())

When this line executes:

async with websockets.serve(handler, "localhost", 8000):

The library opens a TCP socket on the specified host and port and waits for incoming clients. When one connects, it creates a connection object and passes it into your handler function.

The handler is required because it defines what the server does with each connection. The host and port arguments are also important. Both default to None – passing neither raises an error because the OS cannot bind a network server without a port.

You could pass port=0 to let the OS assign a free port automatically, but then you'd need an extra step to figure out which port was chosen, so the client can connect:

server.sockets[0].getsockname()

It’s simpler to specify both host and port explicitly, so the client knows exactly where the server is running.

Set Up the Client

Create client.py in the same folder and add this:

import asyncio
import websockets

async def client():
    async with websockets.connect("ws://localhost:8000") as websocket:
        await websocket.send("Hello server!")
        response = await websocket.recv()
        print("Server replied:", response)

asyncio.run(client())

Test the Connection

First, open a terminal and run server.py. You should see:

Server running at ws://localhost:8000

In a second terminal, run client.py. Messages should appear in both terminals confirming that the connection is active and both sides are communicating.

Note that the server must be running before you start the client – otherwise the client has nothing to connect to, and the connection will fail.

Keeping the server alive: a note on asyncio.Future()

In server.py, there’s a line currently commented out:

await asyncio.Future()

This keeps the server running indefinitely. For local development and testing however, await asyncio.sleep(30) is a simpler alternative. It keeps the server alive for a fixed period without running forever.

File Transfer Over WebSockets

WebSockets aren't limited to text. They support raw bytes too, which means you can send files directly over the connection. Here’s how a client can send a file to a server over a WebSocket connection:

Update server.py

async def file_handler(ws):
    print("Client connected, waiting for file...")
    file_bytes = await ws.recv()  # receive bytes
    with open("received_file.png", "wb") as f:
        f.write(file_bytes)
    print("File received and saved!")
    await ws.send("File received successfully!")

async def main():
    async with websockets.serve(file_handler, "localhost", 8000):
        print("Server running on ws://localhost:8000")
        await asyncio.sleep(50)  # keep server alive

asyncio.run(main())

The handler waits for incoming bytes with await ws.recv(); the websockets library automatically detects whether the incoming message is text or bytes, so no extra configuration is needed. Once received, the file is written to disk in binary mode ("wb") and the server sends a confirmation message back to the client.

Update client.py

import asyncio
import websockets

async def send_file():
    uri = "ws://localhost:8000"
    async with websockets.connect(uri) as ws:
        with open("portfolio-image.png", "rb") as f:  #open file in binary mode
            file_bytes = f.read()
        await ws.send(file_bytes)  # send bytes
        response = await ws.recv()
        print("Server response:", response)

asyncio.run(send_file())

The client opens the image in binary mode ("rb"), reads the entire file into memory as bytes, and sends it in a single ws.send() call. It then waits for the server's confirmation before closing the connection.

Test it

Add an image to your project folder and make sure the filename in client.py matches. Run server.py first, then client.py in a second terminal.

Once the transfer completes, the server saves the file as received_file.png in the same directory. You should see it appear in your workspace immediately.

This approach loads the entire file into memory before sending. For large files, it’s better to read and send them in chunks. But this is the easiest way to understand WebSocket byte transfer.

How to Connect to an External WebSocket

So far you've been connecting to servers you built yourself. But WebSocket clients can also connect to public servers. For example, a client can connect to Postman’s echo server:

import asyncio
import websockets

async def connect_external():
    uri = "wss://ws.postman-echo.com/raw"  # public WebSocket server
    async with websockets.connect(uri) as ws:
        print("Connected to external server!")

        # Send a message
        await ws.send("Hello external server!")
        print("Message sent")

        # Receive response
        response = await ws.recv()
        print("Received from server:", response)
asyncio.run(connect_external())

Notice the client connects to Postman’s echo server using the wss:// URI scheme instead of ws://. This indicates the connection is encrypted using TLS, similar to how https:// secures regular web requests.

An echo server returns exactly what you send it. So "Hello external server!" comes straight back as the response. It's a useful sandbox for testing your client-side WebSocket code without needing your own server.

WebSockets in FastAPI

FastAPI provides a WebSocket object (via Starlette under the hood) to manage real-time connections. You can define WebSocket endpoints just like HTTP routes, while Uvicorn handles the event loop – no manual asyncio server management needed. This makes FastAPI a natural fit for real-time projects, from chat apps to live dashboards and data feeds.

Before jumping into code, here's a quick reference of the core methods you'll be working with.

Accepting:

• await websocket.accept(): the accept() method must be called first, before anything else. Skip it and the connection gets rejected.

Sending:

• await websocket.send_text(data): sends a string.

• await websocket.send_bytes(data): sends binary data.

• await websocket.send_json(data): serializes and sends JSON.

Receiving:

• await websocket.receive_text(): waits for a text message.

• await websocket.receive_bytes(): waits for binary data.

• await websocket.receive_json(): receives and deserializes JSON.

• async for msg in websocket.iter_text(): iterates over incoming messages, exits cleanly on disconnect.

Closing:

• await websocket.close(code=1000): standard code for a normal closure. It accepts an optional “reason” argument.

Here's what the WebSocket lifecycle looks like in FastAPI:

Building a Simple Echo Server with FastAPI

As you saw with the Postman example, an echo server sends back the message a client provides. Let's build one with FastAPI.

1. Install FastAPI:

pip install "fastapi[standard]"

2. Update server.py:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_text()
   
    await websocket.send_text(f"You said: {data}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

A few things to note here compared to the plain websockets library:

• WebSocket endpoints are defined with @app.websocket("/ws") just like an HTTP route.

• await websocket.accept() is required before anything else. FastAPI won't accept connections without it.

• Uvicorn handles the event loop and server startup for you via the if name == "__main__" block. No asyncio.run() or asyncio.Future() needed.

3. Update client.py:

async def test_client():
    uri = "ws://127.0.0.1:8000/ws"
    async with websockets.connect(uri) as ws:
        await ws.send("Hello FastAPI server!")
        response = await ws.recv()
        print("Server replied:", response)

asyncio.run(test_client())

Since the FastAPI server isn't secured with TLS, the client URI uses ws:// instead of wss://. Make sure to match the host and port from your server code.

4. Interact with the echo server:

Start server.py, then run client.py in another terminal. The server terminal should show the echoed message.

How to Handle WebSocket Disconnections in FastAPI

Clients will inevitably disconnect in real-time applications, sometimes intentionally, sometimes unexpectedly. If not handled properly, this can crash your server or leave it in a broken state.

The WebSocketDisconnect exception in FastAPI is raised whenever a client unexpectedly closes the connection, allowing the server to handle disconnects gracefully, log the event, and clean up resources without crashing.

Here’s an example:

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
   
            if "bye" in data or "quit" in data:
                await ws.send_text("Closing connection")
                await ws.close(code=1000, reason="Server requested close")  
                break
            await ws.send_text(f"I got your request: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")  # connection already closed

The server runs a continuous loop waiting for messages. If the client message contains "bye" or "quit", the server responds, calls await ws.close(code=1000), and breaks out of the loop cleanly.

But if the client disconnects unexpectedly, WebSocketDisconnect is caught by the except block and the server moves on without crashing. At this point the connection is already closed on the client side, so calling ws.close() inside the except block is unnecessary.

Conclusion

WebSockets make real-time communication possible by keeping a persistent connection open between client and server. Starting with Python’s websockets library helps clarify how the protocol works under the hood, while frameworks like FastAPI provide the structure needed for production applications.

The parts that trip most people up early on are asyncio and FastAPI's explicit websocket.accept(). With asyncio, the question is usually why it's needed and why the server dies instantly without something keeping it alive. And it's easy to ignore websocket.accept() if you're coming from the plain websockets library where that happens automatically. Once those click, everything else follows naturally.

If you've been working with FastAPI, you've likely encountered dependencies in action. Perhaps you saw Depends() in a tutorial or the docs and were confused for a minute. I certainly was. That confusion sparked weeks of experimenting with this system. The truth is, you can't avoid dependency injection when building backend services with FastAPI. It's baked into the framework's DNA, powering everything from authentication and database connections to request validation.

FastAPI's docs describe its dependency injection system as 'powerful but intuitive.' That’s accurate, once you understand how it works. This article breaks it down, covering function dependencies, class dependencies, dependency scopes, as well as practical examples.

Table of Contents

• Prerequisites

• Dependencies and Dependency Injection in FastAPI

• Getting Started: Environment Setup

• Types of Dependencies in FastAPI

  • How to Use Function Dependencies in FastAPI

  • How to Use Class Dependencies in FastAPI

• Dependency Scope

  • Path Operation Level

  • Router Level

  • Application Level

• Common Use Cases for Dependency Injection

• Conclusion

Prerequisites

To follow along with this article, you should have:

• Working knowledge of Python.

• Ability to create and activate virtual environments.

• Basic understanding of FastAPI.

• Familiarity with Object-Oriented Programming (OOP) concepts.

Dependencies and Dependency Injection in FastAPI

A dependency is a reusable piece of logic, like authentication, database connection, or validation, that your path operations require. Dependency injection (DI) is how FastAPI delivers these dependencies to specific parts of your application: you declare them using Depends() and FastAPI automatically executes them when the associated route receives a request.

Think of it as requesting the tools your application needs. You declare dependencies once and FastAPI provides them wherever needed, with no repetitive setup across routes.

This makes for modular, scalable applications. Without DI, you would have to repeat the same setup code on every endpoint, making updates tedious and bugs more likely.

Getting Started: Environment Setup

Let's set up your development environment to work through the examples in this guide.

Start by creating a project folder, then:

Create and activate a virtual environment:

python -m venv deps
source deps/bin/activate          #on Mac
deps\Scripts\activate             # On Windows

Install FastAPI with all dependencies:

pip install 'fastapi[all]'

Organize your project as follows:

fastapi-deps/
├── deps/                 # Virtual environment
├── function_deps.py
├── class_deps.py
├── router_deps.py
├── app.py
└── requirements.txt

Types of Dependencies in FastAPI

In FastAPI, a dependency is a callable object that retrieves or verifies information before a route executes. Dependencies can be implemented as either functions or classes.

Function dependencies are the most straightforward approach and work well for most use cases, including validation, authentication, and data retrieval. Class dependencies can handle the same tasks but are particularly useful when you need stateful logic, multiple instances with different configurations, or prefer object-oriented patterns.

How to Use Function Dependencies in FastAPI

A function dependency is a helper function (such as for authentication or data retrieval) that can be injected into path operations. To demonstrate, we'll create a simple user authentication dependency using an in-memory database—a list of dictionaries.

Recall the folder structure from earlier? We’ll write this code in fastapi-deps/function_deps.py.

Start by importing the required modules:

from fastapi import FastAPI, Depends, HTTPException
import uvicorn

You bring in FastAPI to create the app instance, Depends for dependency injection, and HTTPException to handle errors gracefully. uvicorn will be used to run the application later.

Next, instantiate the FastAPI application:

app = FastAPI()

app = FastAPI() creates your application instance: the object that will hold all your endpoints and dependencies.

Next, create an in-memory database. Define a list of dictionaries to act as your temporary database. Each dictionary represents a user entry containing a name and a password.

users = [
    {"name": "Ore", "password": "jkzvdgwya12"},
    {"name": "Uche", "password": "lga546"},
    {"name": "Seke", "password": "SK99!"},
    {"name": "Afi", "password": "Afi@144"},
    {"name": "Sam", "password": "goTiger72*"},
    {"name": "Ozi", "password": "xx%hI"},
    {"name": "Ella", "password": "Opecluv18"},
    {"name": "Claire", "password": "cBoss@14G"},
    {"name": "Sena", "password": "SenDaBoss5"},
    {"name": "Ify", "password": "184Norab"}  
]

Then, define a dependency function for user validation. The simple helper function below checks whether a username and password provided by the user match an existing user in the database.

#the dependency function
def user_dep(name: str, password: str):
    for u in users:
        if u["name"] == name and u["password"] == password:
            return {"name": name, "valid": True}

This function expects two string parameters, name and password, from the incoming request. If it finds a match in the users database, it returns a dictionary confirming the user’s validity. FastAPI automatically converts this dictionary into a JSON response.

Next, inject the dependency into a path function:

#the web endpoint
@app.get("/users/{user}")
def get_user(user = Depends(user_dep)) -> dict:
    if not user:
        raise HTTPException(status_code=401, detail="Invalid username or password")
    return user

The user_dep function is injected into the path operation using Depends(). When an HTTP request is made to this endpoint, FastAPI executes the dependency first, validates the input, and passes its return value to the user parameter.

The -> dict: annotation indicates that the function returns a dictionary, which FastAPI auto-converts to JSON. If no matching record is found, an HTTPException with a 401 status code is raised; otherwise, the verified user data is returned.

Now you’ll start the FastAPI server. To start the server, open your terminal in the project directory and run:

uvicorn function_deps:app --reload

• function_deps is the name of your Python file (without the .py extension).

• --reload automatically restarts the server whenever you save changes.

Once it starts, you’ll see an output similar to the image below:

Now you can test the endpoint. Open your browser or the Postman desktop app to validate the user “Seke”. Paste this URL into your browser: http://127.0.0.1:8000/users/{user}?name=Seke&password=SK99!

Alternatively, you can test the endpoint using FastAPI’s built-in docs at: http://127.0.0.1:8000/docs

In the Swagger UI:

• Click on the Get User endpoint

• Click Try it out

• Enter “Seke” in the name field and “SK99!” in the password field

• Click Execute

You should get a 200 status code, with the payload in this image:

You can also test the endpoint with usernames or passwords that don’t exist in the database. Each time, you should see a 401 error like this:

How to Use Class Dependencies in FastAPI

While functions are the most common way to define dependencies, FastAPI also supports class-based dependencies. Classes are useful when you need reusable instances with configurable state or prefer object-oriented patterns.

Class dependencies inject the same way: through the Depends function in your path operation.

Let's convert the user_dep function dependency to a class. It will authenticate users, grant access to valid credentials, and raise exceptions for unauthorized attempts. We'll apply it to a user dashboard endpoint to ensure only authenticated users access their resources.

#Dependency class for user authentication
class UserAuth():
    def __init__(self, name: str, password: str):
        self.name = name
        self.password = password

    def __call__(self):
        #check if name and password entered correspond to any row in the db
        for user in users:
            if user["name"] == self.name and user["password"] == self.password:
                pass
        #If no match found, raise an error
        raise HTTPException(status_code=401, detail="Invalid username or password")

The __init__ method receives the parameters from the request (name and password) and stores them as instance attributes. These can then be accessed in the __call__ method, which contains the dependency logic.

Note that __call__ doesn't return a value in this example. It simply raises an HTTPException if authentication fails. The __call__ method makes the class instance callable, allowing FastAPI to invoke it like a regular function.

Here’s how to inject UserAuth into a path function:

#Injecting the class dependency into a path operation
@app.get("/user/dashboard")
def get_dashboard(user: UserAuth = Depends(UserAuth)):
    return {"message": f"Access granted to {user.name}"}

What's happening here:

When a client requests the /user/dashboard endpoint, FastAPI executes the dependency first. Recognizing UserAuth as a class, FastAPI automatically creates an instance and populates it with values from the query parameters.

Here’s the execution flow to help you understand:

• Depends(UserAuth) tells FastAPI: “Before running this route, create a UserAuth instance.”

• FastAPI extracts name and password from the request URL (for example, /user/dashboard?name=Seke&password=SK99!).

• It then calls UserAuth(name=”Seke”, password=”SK99!”) to create the instance.

• The UserAuth instance, with its stored name and password attributes, is passed to the user parameter in get_dashboard.

• The route function can access user.name and user.password directly.

• If __call__ raises an exception, the route never executes.

Test the endpoint with valid credentials from the users list, and you should see output like this:

A closer look at FastAPI’s official documentation provides an alternative approach to classes as dependencies. However, using the __call__ method, in my opinion, is the most straightforward and self-contained approach. It keeps your authentication logic modular without adding extra code to the path operation.

The trade-off is that class dependencies are more verbose than helper functions, but cleaner for complex logic.

Dependency Scope

FastAPI offers two ways to inject dependencies into a path operation: as a function parameter or via the path decorator. When you include a dependency as a function parameter, the dependency's return value is available within the function. But when injected into the decorator, the dependency executes without passing a return value to the path function.

Beyond single endpoints, FastAPI lets you inject dependencies at the router or global level. Let’s examine these scopes in more detail.

Path Operation Level

While the first example injected dependencies into path function parameters, you can also inject them directly into the decorator using the dependencies parameter. This approach is useful for side-effects (for example, authentication guards, rate limiting or request logging) where the return data is not required in the path operation.

Replace the previous code in fastapi-deps/function_deps.py with this:

#dep function to pass in decorator
def user_dep(name: str, password: str):
    for u in users:
        if u["name"] == name and u["password"] == password:
            return
    raise HTTPException(status_code=401, detail="Invalid username or password")

#path function
@app.get("/users/{user}", dependencies=[Depends(user_dep)])
def get_user() -> dict:
    return {"message" : "Access granted!"}

This decorator-based dependency acts as a pre-check before the endpoint executes. It validates credentials without passing any values to the path function. On authentication failure, FastAPI raises an HTTPException and prevents the path operation from running.

If you test this using a valid name and password from the in-memory database, your output should look like this:

Router Level

Injecting dependencies at the router level allows multiple endpoints to share common logic without repeating the dependency in each route.

We'll use the same user_dep function but inject it at the router level. Add these imports to fastapi-deps/router_deps.py:

from fastapi import APIRouter, Depends

#import the dependency function
from function_deps import user_dep

Then, create an APIRouter instance, passing your dependency to the dependencies parameter. This makes the dependency run automatically for every route you define under this router.

In this example, user_dep executes before get_user() and any other endpoints you add to the router, eliminating the need to declare it on each route.

router = APIRouter(prefix="/users", dependencies=[Depends(user_dep)])

#define the routes with or without additional dependencies
@router.get("/{user}")
def get_user() -> dict:
    return {"message" : "Access granted!"}

In your main application file (app.py), import the router and register it with your FastAPI application using include_router(). This makes all routes defined in the router accessible through your application.

from fastapi import FastAPI
import uvicorn
from router_deps import router as user_router

app = FastAPI()
app.include_router(user_router)

if __name__ == "__main__":
    uvicorn.run("app:app", reload=True)

Start your server and test the route using a valid name–password pair from the users list, then try a mismatched one. You should get a 200 status for the correct credentials and 401 for invalid ones.

Application Level

Application-level dependencies (also called global dependencies) are defined when instantiating the FastAPI app and apply to every route in your application. Unlike router-level dependencies that target specific endpoint groups, app-level dependencies extend across the entire application. Any dependency injected into the FastAPI app object will automatically execute for all path functions.

Let's inject a simple logging dependency alongside the user authentication dependency we've used throughout this article.

Update fastapi-deps/app.py with this code:

from fastapi import FastAPI, Depends
import uvicorn
from function_deps import user_dep
from router_deps import router as user_router
from datetime import datetime

#Basic logging dependency
def log_request():
    print(f"[{datetime.now()}] Request received.")

app = FastAPI(dependencies=[Depends(log_request), Depends(user_dep)])
app.include_router(user_router)

@app.get("/home")
def get_main():
    return "Welcome back!!!"


if __name__ == "__main__":
    uvicorn.run("app:app", reload=True)

When you send a request to any endpoint within this application, log_request acknowledges it and outputs what time the request was made. Since we aren’t sending the logs to any database in particular, it will just print to the terminal (or console) like so:

Request the endpoint with valid credentials using your browser, cURL, Postman, or the Swagger UI. You should get this response:

Common Use Cases for Dependency Injection

Dependency injection solves several common challenges in API development. Here are the most frequent use cases where you'll apply this pattern.

1. Database Connections: Reusing connection logic across multiple endpoints prevents connection leaks, and ensures each request has an isolated session.

2. Authentication & Authorization: Dependencies help validate tokens and verify user roles across protected routes.

3. Logging & Monitoring: A logging dependency can automatically record each request to your monitoring system or database. It is beneficial for debugging and tracking API usage.

4. Rate Limiting: You can control request frequency and prevent API abuse by injecting rate-limiting logic in path functions.

5. Configuration & Settings: FastAPI’s dependency injection system simplifies configuration management by letting you inject settings such as API keys or environment variables wherever needed, keeping your code consistent.

6. Pagination & Filtering: Injecting common parameters like page_size and limit standardize data retrieval patterns across endpoints.

Conclusion

FastAPI's dependency injection system helps you manage shared logic and resources efficiently while adhering to DRY principles. However, knowing when to inject a dependency versus when to skip it is a skill that comes with practice.

Dependency injection isn't needed for simple, standalone logic. But for resources requiring lifecycle management, shared logic, or modularity, FastAPI's dependency injection system simplifies checks and app operations—with or without return values.

This tutorial demonstrates how to clean messy JSON and export the results into a new file, based on a predefined schema. The JSON file we’ll be cleaning contains a dataset of 200 synthetic customer records.

In this tutorial, we’ll apply two methods for cleaning the input data:

• With pure Python

• With pandas

You can apply either of these in your code. But the pandas method is better for large, complex data sets. Let’s jump right into the process.

Here’s what we’ll cover:

• Prerequisites

• Add and Inspect the JSON File

• Define the Target Schema

• How to Clean JSON Data with Pure Python

• How to Clean JSON Data with Pandas

• How to Validate the Cleaned JSON

• Pandas vs Pure Python for Data Cleaning

Prerequisites

To follow along with this tutorial, you should have a basic understanding of:

• Python dictionaries, lists, and loops

• JSON data structure (keys, values, and nesting)

• How to read and write JSON files with Python’s json module

Add and Inspect the JSON File

Before you begin writing any code, make sure that the .json file you intend to clean is in your project directory. This makes it easy to load in your script using the file name alone.

You can now inspect the data structure by viewing the file locally or loading it in your script, with Python’s built-in json module.

Here’s how (assuming the file name is “old_customers.json”):

This shows you whether the JSON file is structured as a dictionary or a list. It also prints out the entire file in your terminal. Mine is a dictionary that maps to a list of 200 customer entries. You should always open up the raw JSON file in your IDE to get a closer look at its structure and schema.

Define the Target Schema

If someone asks for JSON data to be cleaned, it probably means that the current schema is unsuitable for its intended purpose. At this point, you want to be clear on what the final JSON export should look like.

JSON schema is essentially a blueprint that describes:

• required fields

• field names

• data type for each field

• standardized formats (for example, lowercase emails, trimmed whitespace, etc.)

Here’s what the old schema versus the target schema looks like:

As you can see, the goal is to delete the ”customer_id” and ”address” fields in each entry and rename the rest from:

• ”name” to ”full_name”

• ”email” to ”email_address”

• ”phone” to ”mobile”

• ”membership_level” to ”tier”

The output should contain 4 response fields instead of 6, all renamed to fit the project requirements.

How to Clean JSON Data with Pure Python

Let’s explore using Python’s built-in json module to align the raw data with the predefined schema.

Step 1: Import json and time modules

Importing json is necessary because we’re working with JSON files. But we’ll use the time module to track how long the data cleaning process takes.

import json
import time

Step 2: Load the file with json.load()

start_time = time.time()
with open('old_customers.json') as file:
    crm_data = json.load(file)

Step 3: Write a function to loop through and clean each customer entry in the dictionary

def clean_data(records):
    transformed_records = []
    for customer in records["customers"]:
        transformed_records.append({
                "full_name": customer["name"],
                "email_address": customer["email"],
                "mobile": customer["phone"],
                "tier": customer["membership_level"],

                })
    return {"customers": transformed_records}

new_data = clean_data(crm_data)

clean_data() takes in the original data (temporarily) stored in the records variable, transforming it to match our target schema.

Since the JSON file we loaded is a dictionary containing a ”customers” key, which maps to a list of customer entries, we access this key and loop through each entry in the list.

In the for loop, we rename the relevant fields and store the cleaned entries in a new list called ”transformed_records”.

Then, we return the dictionary, with the ”customers” key intact.

Step 4: Save the output in a .json file

Decide on a name for your cleaned JSON data and assign that to an output_file variable, like so:

output_file = "transformed_data.json"
with open(output_file, "w") as f:
    json.dump(new_data, f, indent=4)

You can also add a print() statement below this block to confirm that the file has been saved in your project directory.

Step 5: Time the data cleaning process

At the beginning of this process, we imported the time module to measure how long it takes to clean up JSON data using pure Python. To track the runtime, we stored the current time in a start_time variable before the cleaning function, and we’ll now include an end_time variable at the end of the script.

The difference between the end_time and start_time values gives you the total runtime in seconds.

end_time = time.time()
elapsed_time = end_time - start_time

print(f"Transformed data saved to {output_file}")
print(f"Processing data took {elapsed_time:.2f} seconds")

Here’s how long the data cleaning process took with the pure Python approach:

How to Clean JSON Data with Pandas

Now we’re going to try achieving the same results as above, using Python and a third-party library called pandas. Pandas is an open-source library used for data manipulation and analysis in Python.

To get started, you need to have the Pandas library installed in your directory. In your terminal, run:

pip install pandas

Then follow these steps:

Step 1: Import the relevant libraries

import json
import time
import pandas as pd

Step 2: Load file and extract customer entries

Unlike the pure Python method, where we simply indexed the key name ”customers” to access the list of customer data, working with pandas requires a slightly different approach.

We must extract the list before loading it into a DataFrame because pandas expects structured data. Extracting the list of customer dictionaries upfront ensures that we isolate and clean the relevant records alone, preventing errors caused by nested or unrelated JSON data.

start_time = time.time()
with open('old_customers.json', 'r') as f:
    crm_data = json.load(f)

#Extract the list of customer entries
clients = crm_data.get("customers", [])

Step 3: Load customer entries into a DataFrame

Once you’ve got a clean list of customer dictionaries, load the list into a DataFrame and assign said list to a variable, like so:

#Load into a dataframe
df = pd.DataFrame(clients)

This creates a tabular or spreadsheet-like structure, where each row represents a customer. Loading the list into a DataFrame also allows you to access pandas’ powerful data cleaning methods like:

• drop_duplicate(): removes duplicate rows or entries from a DataFrame

• dropna(): drops rows with any missing or null data

• fillna(value): replaces all missing or null data with a specified value

• drop(columns): drops unused columns explicitly

Step 4: Write a custom function to rename relevant fields

At this point, we need a function that takes in a single customer entry – a row – and returns a cleaned version that fits the target schema (“full_name”, “email_address”, “mobile” and “tier”).

The function should also handle missing data by setting default values like ”Unknown” or ”N/A” when a field is absent.

P.S: At first, I used drop(columns) to explicitly remove the “address” and “customer_id” fields. But it’s not needed in this case, as the transform_fields() function only selects and renames the required fields. Any extra columns are automatically excluded from the cleaned data.

Step 5: Apply schema transformation to all rows

We’ll use pandas' apply() method to apply our custom function to each row in the DataFrame. This will creates a Series (for example, 0 → {...}, 1 → {...}, 2 → {...}), which is not JSON-friendly.

As json.dump() expects a list, not a Pandas Series, we’ll apply tolist(), converting the Series to a list of dictionaries.

#Apply schema transformation to all rows
transformed_df = df.apply(transform_fields, axis=1)

#Convert series to list of dicts
transformed_data = transformed_df.tolist()

Another way to approach this is with list comprehension. Instead of using apply() at all, you can write:

transformed_data = [transform_fields(row) for row in df.to_dict(orient="records")]

orient=”records” is an argument for df.to_dict that tells pandas to convert the DataFrame to a list of dictionaries, where each dictionary represents a single customer record (that is, one row).

Then the for loop iterates through every customer record on the list, calling the custom function on each row. Finally, the list comprehension ([...]) collects the cleaned rows into a new list.

Step 6: Save the output in a .json file

#Save the cleaned data
output_data = {"customers": transformed_data}
output_file = "applypandas_customer.json"
with open(output_file, "w") as f:
    json.dump(output_data, f, indent=4)

I recommend picking a different file name for your pandas output. You can inspect both files side by side to see if this output matches the result you got from cleaning with pure Python.

Step 7: Track runtime

Once again, check for the difference between start time and end time to determine the program’s execution time.

end_time = time.time()
elapsed_time = end_time - start_time

#print(f"Transformed data saved to {output_file}")
print(f"Transformed data saved to {output_file}")
print(f"Processing data took {elapsed_time:.2f} seconds")

When I used list comprehension to apply the custom function, my script’s runtime was 0.03 seconds, but with pandas’ apply() function, the total runtime dropped to 0.01 seconds.

Final output preview:

If you followed this tutorial closely, your JSON output should look like this – whether you used the pandas method or the pure Python approach:

How to Validate the Cleaned JSON

Validating your output ensures that the cleaned data follows the expected structure before being used or shared. This step helps to catch formatting errors, missing fields, and wrong data types early.

Below are the steps for validating your cleaned JSON file:

Step 1: Install and import jsonschema

jsonschema is a third-party validation library for Python. It helps you define the expected structure of your JSON data and automatically check if your output matches that structure.

In your terminal, run:

pip install jsonschema

Import the required libraries:

import json
from jsonschema import validate, ValidationError

validate() checks whether your JSON data matches the rules defined in your schema. If the data is valid, nothing happens. But if there’s an error – like a missing field or wrong data type – it raises a ValidationError.

Step 2: Define a schema

As you know, JSON schema changes with each file structure. If your JSON data differs from what we’ve been working with so far, learn how to create a schema here. Otherwise, the schema below defines the structure we expect for our cleaned JSON:

schema = {
    "type": "object",
    "properties": {
        "customers": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "full_name": {"type": "string"},
                    "email_address": {"type": "string"},
                    "mobile": {"type": "string"},
                    "tier": {"type": "string"}
                },
                "required": ["full_name", "email_address", "mobile", "tier"]
            }
        }
    },
    "required": ["customers"]
}

• The data is an object that must contain a key: "customers".

• "customers" must be an array (a list), with each object representing one customer entry.

• Each customer entry must have four fields–all strings:

  • "full_name"

  • "email_address"

  • "mobile"

  • "tier"

• The "required" fields ensure that none of the relevant fields are missing in any customer record.

Step 3: Load the cleaned JSON file

with open("transformed_data.json") as f:
    data = json.load(f)

Step 4: Validate the data

For this step, we’ll use a try. . . except block to end the process safely, and display a helpful message if the code raises a ValidationError.

try:
    validate(instance=data, schema=schema)
    print("JSON is valid.")
except ValidationError as e:
    print("JSON is invalid:", e.message)

Pandas vs Pure Python for Data Cleaning

From this tutorial, you can probably tell that using pure Python to clean and restructure JSON is the more straightforward approach. It is fast and ideal for handling small datasets or simple transformations.

But as data grows and becomes more complex, you might need advanced data cleaning methods that Python alone does not provide. In such cases, pandas becomes the better choice. It handles large, complex datasets effectively, providing built-in functions for handling missing data and removing duplicates.

You can study the Pandas cheatsheet to learn more data manipulation methods.