DEV Community: Rajat Bansal

Stop Swallowing Errors: A Guide to Professional Exception Handling in APIs

Rajat Bansal — Sun, 21 Dec 2025 18:58:41 +0000

We all know try-except (or try-catch). It’s one of the first things we learn to stop our programs from crashing.

However, I often see fresher developers wrapping massive chunks of code in try-except blocks just for the sake of using them, often swallowing errors with a pass. While this keeps the server running, it creates unmaintainable code and hides critical bugs.

We need to be very careful about where we use it and where not to.

In this article, I will discuss the correct architecture for exception handling, specifically in the context of an ICR (Intelligent Character Recognition) API—a service that takes images or PDFs and extracts text from them. An API generally consists of two main parts: the Core Logic (the heavy lifting) and the API Route (the interface).

1. When NOT to use Try-Except: Core Logic vs. Coding Logic

The most important rule: Never use try-except for coding logic that can be handled with a simple if condition.

A common mistake is using exception handling to manage predictable data structure issues, like index-out-of-range errors when iterating through lists.

Suppose your ICR engine returns a list of words, and you assume that the value you want always follows the word 'Address'.

The Wrong Way (Lazy):
You might "try" to access the next index and just ignore it if it fails.

data = ["Name:", "John", "Address:"] # Address value is missing!
n = len(data)

try:
    # We assume the next index exists. If it doesn't, it crashes into the except block.
    for i in range(n):
      if data[i] == 'Address:':
          address = data[i+1] 
except IndexError:
    # Swallowing the error. The code proceeds, but 'address' might be undefined.
    pass

The Right Way (Logic):
If you can predict an error, write logic for it. Don't rely on the exception mechanism.

data = ["Name:", "John", "Address:"]
n = len(data)

for i in range(n):
  if data[i] == 'Address:':
      # Logic check: Ensure i+1 actually exists before accessing it
      if i + 1 < n:
         address = data[i+1]
      else:
         # Handle the missing data case gracefully without an exception
         address = None
         print("Found 'Address:' label but no value followed it.")

In the second example, we control the flow with logic. It's clearer and faster.

2. Where we DO need Try-Except

We genuinely need try-except when dealing with things we cannot control—external dependencies.

In our ICR API context, this includes:

Calling an external OCR microservice.
Connecting to a database to log requests.
Reading a physical file from S3 or local disk.

These external systems can fail at any time for reasons unrelated to your code. This is where you must catch exceptions.

3. Designing the API: The "Wrong Way" to handle business rules

Now, let's talk about API architecture. Suppose our ICR API has a business rule: We do not accept PDFs longer than 10 pages because processing them takes too long.

A very common mistake junior developers make is handling these "business errors" in the core logic and returning string messages indicating failure.

Let's look at why this approach is messy.

The Core Logic (Bad Approach):

# --- core_icr_logic.py ---

def extract_data_from_pdf(pdf_file):
    # 1. Check page count constraint
    page_count = get_pdf_page_count(pdf_file)

    if page_count > 10:
        # THE BAD PRACTICE: Returning an error string
        return "Error: PDF has too many pages. Limit is 10."

    # 2. External Dependency: Call the heavy ICR engine
    try:
        extracted_text = call_heavy_icr_engine(pdf_file)
        return extracted_text
    except ConnectionError:
        # THE BAD PRACTICE: Returning another error string
        return "Error: ICR Engine is down."

The API Route (Bad Approach):

Now look at how ugly the API route becomes. It has to inspect the returned string to figure out if it succeeded or failed.

# --- api_routes.py ---

@app.post("/extract")
def extract_api(file: UploadFile):
    result = extract_data_from_pdf(file)

    # The route has to guess what happened based on string matching. This is fragile.
    if isinstance(result, str) and result.startswith("Error: PDF has too many"):
         return {"msg": result}, 422 # Unprocessable Entity

    elif isinstance(result, str) and result.startswith("Error: ICR Engine"):
         return {"msg": result}, 503 # Service Unavailable

    # Assuming it's success if it's not an error string... dangerous assumption.
    return {"data": result}, 200

Why is this bad?

Hard to maintain: If you change the error string text in the core logic, you break the API route check.
Mixed Concerns: The core logic shouldn't care about HTTP responses, and the route shouldn't have to parse strings to understand business rules.

4. The "Right Way": Custom Exceptions and Route Handling

The cleanest way to handle this is to separate concerns:

Core Logic: Fails loudly by raising exceptions when business rules are violated.
API Route: Is the only place that catches exceptions to determine the HTTP status code.

Step A: Create Custom Exceptions

Define exceptions specific to your business domain.

# --- exceptions.py ---
class PageLimitExceededException(Exception):
    """Raised when the PDF is too long for us to process"""
    pass

class ICREngineDownException(Exception):
    """Raised when the external ICR service is unreachable"""
    pass

Step B: The Core Logic (Raises, never returns errors)

The core logic now only returns successful data. If something is wrong, it raises a specific flag (exception).

# --- core_icr_logic.py ---
from exceptions import PageLimitExceededException, ICREngineDownException

def extract_data_from_pdf(pdf_file):
    # 1. Check page count constraint
    page_count = get_pdf_page_count(pdf_file)

    if page_count > 10:
        # Just raise it and stop immediately. Don't return strings.
        raise PageLimitExceededException("PDF exceeds 10 page limit.")

    # 2. External Dependency
    try:
        extracted_text = call_heavy_icr_engine(pdf_file)
        return extracted_text
    except ConnectionError:
        # Wrap the low-level connection error in our domain exception
        raise ICREngineDownException("Could not connect to OCR service.")

Step C: The API Route (The "Catch-All")

The route becomes incredibly clean. It tries the main operation and then has a specific bucket for every type of failure.

# --- api_routes.py ---
from exceptions import PageLimitExceededException, ICREngineDownException

@app.post("/extract")
def extract_api(file: UploadFile):
    try:
        # The happy path is very clear
        result = extract_data_from_pdf(file)
        return {"status": "success", "data": result}, 200

    except PageLimitExceededException as e:
        # Catch business rule violation -> 422
        return {"status": "error", "msg": str(e)}, 422

    except ICREngineDownException as e:
        # Catch external dependency failure -> 503
        # Log this critical error here as well
        print(f"CRITICAL: {e}") 
        return {"status": "error", "msg": "Service temporarily unavailable"}, 503

    except Exception as e:
        # Catch any unexpected bugs -> 500 generic server error
        print(f"UNEXPECTED ERROR: {e}")
        return {"status": "error", "msg": "Internal Server Error"}, 500

Conclusion

By moving from "returning error strings" to "raising custom exceptions," your ICR API code becomes much easier to read and maintain.

Do not use try-except for logic that if/else can handle.
Never catch an exception in your core business logic just to return an error message. Raise it.
Catch exceptions only at the "entry point" (the API route) to decide which HTTP status code to send back to the client.

Automate Kubernetes Deployments and Upgrades using Python

Rajat Bansal — Sun, 02 Mar 2025 14:37:32 +0000

Hi, I am Rajat, working in Data Science and DevOps.

Being a data scientist, I have a good command over Python. Generally, you will find that a new deployment and upgrade is not an easy job. It includes writing config files and creating complex pipelines.

But wait... then I remembered I know Python. Why am I not using my Python skills to automate things?

So, I started writing some scripts to automate the jobs. First, I wrote some scripts that would help me in my deployments.

When we talk about a Kubernetes deployment, what are the first things we need to do? It can be a plain deployment using YAML files or using the Helm chart. Here's the list:

Create a namespace
Create image-pull secrets
Create a YAML file (Deployment, Service, Ingress, HPA, ConfigMap, Secret) or the Helm file if you are using Helm chart
Finally, deploy and validate

Now, let's start doing this one by one.

import argparse
import yaml
import os
from kubernetes import client, config, watch
from kubernetes.client.rest import ApiException
import base64
from traceback import print_exc
import logging

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

class KubernetesManager:
    def __init__(self):
        logging.info("Initializing Kubernetes manager...")
        config.load_kube_config()
        self.core_api = client.CoreV1Api()
        self.apps_api = client.AppsV1Api()
        self.namespaces = []
        self.list_namespaces()

    def create_namespace(self, namespace_name):
        try:
            logging.info(f"Creating namespace: {namespace_name}")
            if namespace_name in self.namespaces:
                logging.info(f"Namespace '{namespace_name}' already exists.")
                return
            namespace = client.V1Namespace(metadata=client.V1ObjectMeta(name=namespace_name))
            self.core_api.create_namespace(namespace)
            logging.info(f"Namespace '{namespace_name}' created successfully.")
            self.create_image_pull_secret(namespace_name)
        except ApiException as e:
            logging.error(f"Error creating namespace: {e}")

    def create_image_pull_secret(self, namespace_name, secret_name, server, username, password):
        try:
            logging.info(f"Creating image pull secret in namespace: {namespace_name}")
            docker_config_json = {
                "auths": {
                    server: {
                        "username": username,
                        "password": password,
                        "auth": base64.b64encode(f"{username}:{password}".encode()).decode()
                    }
                }
            }
            secret = client.V1Secret(
                metadata=client.V1ObjectMeta(name=secret_name),
                type="kubernetes.io/dockerconfigjson",
                data={".dockerconfigjson": base64.b64encode(json.dumps(docker_config_json).encode()).decode()}
            )
            self.core_api.create_namespaced_secret(namespace_name, secret)
            logging.info(f"ImagePull secret '{secret_name}' created in namespace '{namespace_name}'.")
        except ApiException as e:
            logging.error(f"Error creating secret: {e}")

    def list_namespaces(self):
        try:
            namespaces = self.core_api.list_namespace().items
            self.namespaces = [
                ns.metadata.name
                for ns in namespaces
                if ns.metadata.name not in ["default", "kube-public", "kube-system", "kube-node-lease", "kube-proxy", "kube-service-catalog", "prometheus", "kubernetes-dashboard"]
            ]
        except ApiException as e:
            logging.error(f"Error listing namespaces: {e}")

    def check_rollout_status(self, deployment_name, namespace, timeout_seconds=240):
        try:
            logging.info(f"Checking rollout status for deployment '{deployment_name}' in namespace '{namespace}'...")
            start_time = time.time()

            w = watch.Watch()
            for event in w.stream(
                self.apps_api.list_namespaced_deployment,
                namespace=namespace,
                field_selector=f"metadata.name={deployment_name}",
                timeout_seconds=timeout_seconds,
            ):
                deployment = event['object']
                status = deployment.status

                if (
                    status.updated_replicas == deployment.spec.replicas
                    and status.replicas == deployment.spec.replicas
                    and status.available_replicas == deployment.spec.replicas
                    and status.observed_generation >= deployment.metadata.generation
                ):
                    logging.info(f"Deployment '{deployment_name}' in namespace '{namespace}' has been successfully rolled out.")
                    w.stop()
                    return True

                if time.time() - start_time > timeout_seconds:
                    logging.error(f"Timeout: Deployment '{deployment_name}' in namespace '{namespace}' did not roll out within {timeout_seconds} seconds.")
                    w.stop()
                    return False

                elapsed_time = time.time() - start_time
                logging.info(f"Waiting for deployment '{deployment_name}' to roll out... Time elapsed: {elapsed_time:.2f} seconds")
                logging.info(f"Updated Replicas: {status.updated_replicas}, Available Replicas: {status.available_replicas}")

        except ApiException as e:
            logging.error(f"Error checking rollout status: {e}")
            return False

Let me explain the code above:

We have a class KubernetesManager which has methods to create a namespace and image pull secret.
The create_namespace method creates a namespace if it does not already exist.
The create_image_pull_secret method creates an image pull secret in the specified namespace.
The list_namespaces method lists all the namespaces in the cluster.
The check_rollout_status method monitors the rollout status of a deployment in a namespace.

Suppose you are using the YAML files for deployment or even Helm chart, you can write other functions to do the deployment. Add the below code to the above code.

import subprocess
import argparse

def deploy_using_yaml(yaml_file, namespace):
    logging.info(f"Deploying using yaml file: {yaml_file}")
    command = [
        "kubectl",
        "apply",
        "-f",
        yaml_file,
        "--namespace",
        namespace
    ]
    try:
        subprocess.run(command, check=True)
        logging.info(f"Deployment using yaml file '{yaml_file}' completed successfully.")
    except subprocess.CalledProcessError as e:
        raise Exception(f"Error deploying using yaml file: {e}")

def final_deploy(yaml_file_path, namespace):
    k8s_manager.create_namespace(namespace)
    logging.info(f"Namespace '{namespace}' created successfully.")
    k8s_manager.create_image_pull_secret(namespace, "my-secret", "docker.io", "username", "password")
    logging.info(f"ImagePull secret created successfully.")
    deploy_using_yaml(yaml_file_path, namespace)
    logging.info(f"Deployment using yaml file '{yaml_file_path}' completed successfully.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Deploy an application to Kubernetes.")
    parser.add_argument("--task", required=True, choices=["deploy", "rollback"], help="The task to perform.")
    parser.add_argument("--yaml-file", required=True, help="The path to the yaml file.")
    parser.add_argument("--namespace", required=True, help="The namespace to deploy the application.")
    args = parser.parse_args()

    if args.task == "deploy":
        final_deploy(args.yaml_file, args.namespace)
    elif args.task == "rollback":
        rollback(args.namespace)

If you want to deploy, you can run the following command:

python3 deploy.py --task deploy --yaml-file deployment.yaml --namespace my-namespace

I hope this helps you in automating your Kubernetes deployments using Python. If you have any questions, feel free to ask.

Develop in Multiple OS/Production-like Environments with Docker

Rajat Bansal — Sat, 23 Nov 2024 20:39:24 +0000

In this guide, we will walk through the steps to create a Docker container using a Docker Compose file, attach a volume to your code folder, and develop directly inside the container.

Prerequisites

Docker installed on your machine
Docker Compose installed on your machine

Step 1: Create a Dockerfile

First, create a Dockerfile in your project directory. This file will define the environment for your application.

I will work with a Python application that I will run with Gunicorn. As you know, Gunicorn is not supported on Windows, but using Docker, you can run it on Windows as well.

Let's start writing our Dockerfile. Make sure you keep the Dockerfile in the root of your project directory.

This is how our Dockerfile will look like:

# Use an official Python runtime as a parent image
FROM python:3.10-slim-buster

# I am using python 3.10, you can use any version you want and slim buster is a very light weight image with least vulnerabilities

# Set the working directory to /app, in Dockerfile we specify a working directory, this is the directory where the command will be executed by default

WORKDIR /app

# If you have any requirements, add them here and install them, otherwise you can skip this step

COPY requirements.txt ./

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

# now we shall give a entrypoint to our container, this is the command that will be executed when the container starts or we can give this when we shall start the container

Step 2: Create a Docker Compose File

Next, create a docker-compose.yml file in your project directory. This file will define the services for your application.

services:
  app: # this is the name of the service, you can give any name you want
    build: 
      context: . # this is the path to the Dockerfile, if the Dockerfile is in the root of the project directory, you can use `.` to specify the current directory
      dockerfile: Dockerfile # this is the name of the Dockerfile, if you have named your Dockerfile something else, you can specify it here
    volumes:
      - .:/app # this is most important part, this is the volume mapping, this will map the current directory to the /app directory in the container and the changes you will do inside container will be reflected in the host machine and vice versa
    ports:
      - "8000:8000" # this is the port mapping, you can map any port you want

    entrypoint: tail -f /dev/null # this is the entrypoint, this is the command that will be executed when the container starts, this is a dummy command that will keep the container running and we can develop inside the container

Now we need to build the container, for that we need to run the following command:

docker compose up --build

This will look like this:

Now the main part comes, we will open the container in our vs code. For this we need to install the Remote - Development extensions pack in our vs code.

Now we need to click on the blue icon on the bottom left corner of the vs code and click on Remote-Containers: Attach to Running Container... if you don't find this then click on Dev Container and then click on Attach to Running Container...

There you will find the name of your container, that you specified in docker-compose.yml file, click on that and you will be inside the container , a new window will open and you will be inside the container.

You can see in the left corner the name of the container is written, that means you are inside the container.

Now you can open the /app folder and start developing inside the container, you can run your application inside the container and see the changes in the host machine.

This is like a fresh machine, you can install any dependencies you want, you can run any command you want, you can do anything you want, and the best part is you can do all this in any OS, you can develop in windows and run in linux, you can develop in linux and run in windows, you can develop in mac and run in windows, you can do anything you want.

This is the power of docker, you can create a production like environment in your local machine and develop in that environment, you can test your application in that environment, you can do anything you want.

But make sure that if we stop the container, all the changes except the code will be lost, like if you install any library that will be lost.

How to solve this problem.

If you have to install any python library, make sure you add that to requirements.txt so that it will be installed when you build the container again.

If you have to install any other library, make sure you add that to the Dockerfile so that it will be installed when you build the container again.

For example, if you have to install vim in the container, you can add this line to the Dockerfile:

RUN apt-get update && apt-get install -y vim

You can add this command to the Dockerfile and when you build the container again, this will be installed in the container.

Step 3: Run Your Application Inside the Container

Now that your container is up and running, you can run your application inside the container. For example, if you are using Gunicorn to run a Python application, you can use the following command:

gunicorn --bind 0.0.0.0:8000 wsgi:app

This command will start your application on port 8000 inside the container. You can then access your application by navigating to http://localhost:8000 on your host machine.

Step 4: Stop the Container

To stop the container, you can use the following command:

docker compose down

This command will stop and remove the container, but your code and any changes made to the code will remain intact on your host machine.

This is how you can develop in multiple OS/Production-like environments with Docker.

I hope this guide was helpful to you. If you have any questions or feedback, please feel free to reach out to me.