aniket-tools 0.1.3

Reusable API response and exception handling utilities with FastAPI integration.

aniket_tools

aniket_tools is a reusable Python library for three things:

• building consistent API responses
• translating exceptions into user-friendly messages
• logging raw technical failures for developers

This README only covers library usage and code behavior.

Install

pip install aniket_tools

For local development:

pip install .
pip install ".[full]"

Import Style

Import from the package root only:

from aniket_tools import (
    ApiError,
    ErrorHandler,
    ExceptionHandler,
    PaginationRes,
    create_response,
    explain_error,
    get_logger,
    get_status_code,
    handle_exception,
    logs,
    unified_exception_handler,
    value_correction,
)

Quick Start

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from aniket_tools import create_response, unified_exception_handler

app = FastAPI()

app.add_exception_handler(HTTPException, unified_exception_handler)
app.add_exception_handler(Exception, unified_exception_handler)
app.add_exception_handler(RequestValidationError, unified_exception_handler)


@app.get("/health")
async def health():
    return create_response(200, data={"status": "ok"})

Public API

create_response

Builds the standard success or error payload.

Params:

• response_code: int
• data: Any = None
• schema: Any | None = None
• pagination: Mapping[str, Any] | PaginationRes | None = None
• error_message: str | None = None
• error_code: str | None = None
• details: Sequence[Mapping[str, Any]] | None = None
• meta: Mapping[str, Any] | None = None
• as_json_response: bool = True

Usage:

from aniket_tools import create_response

return create_response(
    200,
    data={"name": "Aniket"},
    pagination={"page": 1, "rows": 10, "total_rows": 100},
    meta={"request_id": "req-1", "trace_id": "trace-1"},
)

Error usage:

return create_response(
    404,
    error_message="Report not found.",
    error_code="report_not_found",
)

Pagination usage with a shared library model:

from aniket_tools import PaginationRes, create_response

return create_response(
    200,
    data=rows,
    pagination=PaginationRes(page=1, rows=25, total_rows=250),
)

Pagination usage with a dictionary:

return create_response(
    200,
    data=rows,
    pagination={
        "page": 1,
        "rows": 25,
        "total_rows": 250,
    },
)

PaginationRes

Shared pagination structure for all projects using the library.

Fields:

• page: int
• rows: int
• total_rows: int

Behavior in create_response(...):

• these three fields are validated as the required core pagination contract
• page must be an integer greater than or equal to 1
• rows must be an integer greater than or equal to 0
• total_rows must be an integer greater than or equal to 0
• extra pagination metadata is still allowed and preserved

If pagination is invalid, create_response(...) returns a 422 validation-style error response instead of returning a broken payload.

meta

meta is included in every JSON response body produced by create_response(...) and exception payloads built by ErrorHandler.

Typical use:

• request_id
• trace_id
• version
• path
• feature or tenant context

Behavior:

• in normal responses, pass meta={...} directly to create_response(...)
• in exception responses, build_payload(...) automatically includes request-derived fields like request_id and path
• 204 No Content is the only exception because it has no response body

value_correction

Normalizes output values before returning them.

What it currently fixes:

• trims strings
• converts Decimal to float
• converts datetime and date to ISO strings
• converts timedelta to string
• rounds floats to 2 decimals
• recursively cleans dict, list, tuple, and set

Params:

• data: Any

Usage:

from decimal import Decimal
from aniket_tools import value_correction

cleaned = value_correction({
    "amount": Decimal("10.50"),
    "name": "  demo  ",
})

logs

Logs a message using the default logger or a provided logger.

Params:

• msg: object = ""
• type: str = "info"
• file_name: str | Path | None = None
• logger: logging.Logger | None = None
• dialect: object | None = None

Supported type values: 7 total

• debug
• info
• warning
• error
• critical
• notset
• query

Unknown type values fall back to info.

query is intended for SQL logging. If msg exposes .compile(), logs(..., type="query") will try to render the SQL with literal_binds=True and log it as INFO. You can pass:

• a dialect name such as "postgresql", "mysql", "sqlite", "mssql", or "oracle"
• a SQLAlchemy engine, async engine, connection, async connection, session, or async session
• a dialect object directly

If no dialect can be resolved, it falls back to generic compilation or str(msg).

Usage:

from aniket_tools import logs

logs("report created", type="info")
logs("database failed", type="error", file_name="logs/app")

Query usage:

from sqlalchemy import select
from aniket_tools import logs

statement = select(User).where(User.id == 7)

logs(statement, type="query", dialect="postgresql")

from aniket_tools import logs

# Sync or async SQLAlchemy bind objects also work.
logs(statement, type="query", dialect=session)
logs(statement, type="query", dialect=engine)

get_logger

Returns a configured Python logger.

Params:

• name: str = "aniket_tools"
• file_name: str | Path | None = None

Usage:

from aniket_tools import get_logger

logger = get_logger("my_app", file_name="logs/app.log")
logger.info("started")

ApiError

Custom business error for controlled API failures.

Params:

• message: str
• status_code: int = 400
• code: str = "api_error"
• details: list[dict[str, Any]] | None = None
• log_message: str | None = None

Usage:

from aniket_tools import ApiError

raise ApiError(
    "Report is not ready.",
    status_code=409,
    code="report_pending",
    details=[{"field": "report_id", "message": "still processing"}],
)

unified_exception_handler

FastAPI exception handler that logs the raw exception and returns the standard JSON error payload.

Params:

• request
• exc: Exception

Usage:

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from aniket_tools import unified_exception_handler

app = FastAPI()

app.add_exception_handler(HTTPException, unified_exception_handler)
app.add_exception_handler(Exception, unified_exception_handler)
app.add_exception_handler(RequestValidationError, unified_exception_handler)

ExceptionHandler

Route-level helper that converts an exception into a FastAPI HTTPException.

Params:

• exc: Exception

Usage:

from aniket_tools import ExceptionHandler

try:
    raise ValueError("invalid meter id")
except Exception as exc:
    ExceptionHandler(exc)

handle_exception

Returns the standard error payload directly. If FastAPI is installed, it returns a JSONResponse; otherwise it returns a plain dictionary.

Params:

• exc: Exception
• request = None
• meta: Mapping[str, Any] | None = None

Usage:

from aniket_tools import handle_exception

payload_or_response = handle_exception(ValueError("invalid input"))

explain_error

Returns the final user-facing message for an exception.

Params:

• exc: Exception

Usage:

from aniket_tools import explain_error

message = explain_error(ValueError("invalid input"))

get_status_code

Returns the final HTTP status code for an exception.

Params:

• exc: Exception

Usage:

from aniket_tools import get_status_code

status_code = get_status_code(ValueError("invalid input"))

ErrorHandler

Core exception translator class. This is the class to study if you want to change library behavior.

Params:

• logger_name: str = "aniket_tools.errors"
• use_default_message_for_long_errors: bool = True

Usage:

from aniket_tools import ErrorHandler

handler = ErrorHandler(use_default_message_for_long_errors=True)
info = handler.describe(ValueError("invalid input"))
payload = handler.build_payload(
    ValueError("invalid input"),
    meta={"trace_id": "trace-1"},
)

Standard Response Shape

Success response

{
  "success": true,
  "response_code": 200,
  "meta": {
    "request_id": "req-1"
  },
  "data": {
    "status": "ok"
  }
}

Error response

{
  "success": false,
  "response_code": 422,
  "error_message": "One or more fields are invalid.",
  "error_type": "RequestValidationError",
  "meta": {
    "request_id": "req-1",
    "path": "/reports"
  },
  "error": {
    "code": "validation_error",
    "type": "RequestValidationError",
    "message": "One or more fields are invalid.",
    "details": [
      {
        "field": "email",
        "message": "field required"
      }
    ]
  },
  "errors": [
    {
      "field": "email",
      "message": "field required"
    }
  ]
}

Why old and new keys exist:

• error_message and error.message are aliases
• errors and error.details are aliases

This keeps older projects working while giving a cleaner nested error object to newer code.

Code Structure

src/aniket_tools/
  __init__.py
  _compat.py
  exceptions.py
  logging.py
  responses.py

What each file owns:

• __init__.py: public export surface
• _compat.py: optional imports and type compaction helpers
• responses.py: response formatting and value normalization
• logging.py: logger creation and logging helpers
• exceptions.py: exception classification, message translation, status mapping, and payload generation

Deep Exception System Knowledge

This section is for future editing.

Full flow

When an exception reaches FastAPI, the path is:

1. FastAPI calls unified_exception_handler(request, exc)
2. unified_exception_handler calls ErrorHandler.log_exception(...)
3. unified_exception_handler calls ErrorHandler.handle_exception(...)
4. handle_exception(...) calls build_payload(...)
5. build_payload(...) calls describe(...)
6. describe(...) decides status code, error code, message, and optional details
7. build_payload(...) merges response meta like request_id, path, or custom trace_id
8. build_payload(...) converts that decision into the final JSON structure

The most important method in the whole library is ErrorHandler.describe(...).

Internal decision objects

exceptions.py uses ErrorInfo as the normalized internal result:

ErrorInfo(
    status_code=422,
    code="validation_error",
    message="One or more fields are invalid.",
    details=[...],
)

That means the code works in two steps:

• step 1: classify any exception into ErrorInfo
• step 2: turn ErrorInfo into API JSON

This separation matters. If you want to change decision logic, edit describe(...). If you want to change JSON shape, edit build_payload(...).

Deep Edit Guide For exceptions.py

1. Common exceptions

Common Python exceptions are handled near the end of ErrorHandler.describe(...).

Current branch:

if isinstance(exc, (ValueError, TypeError, KeyError, IndexError, AssertionError)):
    return ErrorInfo(
        status_code=400,
        code="bad_request",
        message=_safe_client_message(exc, _default_message(400)),
    )

What this means:

• these errors are treated as client-side bad input
• status code is 400
• message is the exception text if it is short and safe
• very long messages fall back to The request data is invalid. by default
• if you create ErrorHandler(use_default_message_for_long_errors=False), long single-line messages can be preserved

Where to edit common exceptions:

• add or remove exception classes in this branch
• change the code from bad_request to another code if needed
• change _safe_client_message(...) behavior if you want stricter or looser exposure of raw messages

2. HTTP exceptions

HTTP exceptions are handled before generic exceptions.

Current logic:

• keep the original status code
• if status is 500 or higher, hide raw detail and use a default safe message
• if status is 400 to 499 and detail is a short string, use it
• if the detail is very long, the default safe message is used unless use_default_message_for_long_errors=False

This is important because server errors should not leak raw internal detail to the client.

Where to edit:

• ErrorHandler.describe(...) in the HTTP_EXCEPTION_TYPES branch
• _clean_message(...) if you want to change what counts as safe user text

3. Validation exceptions

Validation is special because it is not just one message. It also needs field-level detail.

Current validation flow:

• describe(...) detects RequestValidationError or PydanticValidationError
• it returns status 422
• it sets code validation_error
• it calls _normalize_validation_errors(...)

_normalize_validation_errors(...) does three important things:

• reads exc.errors()
• removes framework prefixes like body, query, and path
• turns nested location tuples into a flat field path like user.email

Example input from FastAPI:

{"loc": ("body", "email"), "msg": "field required"}

Normalized output:

{"field": "email", "message": "field required"}

Where to edit validation behavior:

• edit _normalize_validation_errors(...) to change field formatting
• edit the validation branch in describe(...) to change status, code, or top-level message
• edit build_payload(...) if you want validation details under a different payload key

When you should edit validation separately from common errors:

• when frontend needs field-wise messages
• when request body, query, or path errors need custom display
• when nested models should expose a different field path format

4. Database exceptions

Database exceptions are the deepest part of the library.

Current target scope:

• PostgreSQL
• TimescaleDB, which is PostgreSQL-based
• MySQL
• SQLAlchemy-wrapped driver errors from those databases

They are split into categories before ErrorHandler is even used:

• INTEGRITY_EXCEPTION_TYPES
• DATA_EXCEPTION_TYPES
• DB_UNAVAILABLE_EXCEPTION_TYPES
• PROGRAMMING_EXCEPTION_TYPES
• GENERIC_DATABASE_EXCEPTION_TYPES

This is why database logic is cleaner than having one huge if "duplicate key" in str(exc) block.

How database messages are built

_database_message(exc) is the main user-message translator for database failures.

Its flow is:

• prefer exc.orig when SQLAlchemy wraps the real driver error
• convert the message to lowercase for pattern matching
• extract SQL text when available
• try specific parsers before generic fallback text

Current order inside _database_message(...) matters:

1. enum parsing
2. argument type mismatch parsing
3. duplicate key handling
4. foreign key handling
5. not-null handling
6. check constraint handling
7. SQL programming error handling
8. database unavailable handling
9. generic type mismatch heuristics
10. asyncpg-specific fallback
11. final generic database message

That order is intentional. The more specific cases must come first.

Argument type mismatch parsing

_describe_argument_type_error(...) handles errors like:

• invalid input for query argument $2
• expected int, got str

It tries to turn database parameter numbers into real column names.

Example SQL:

INSERT INTO billing (site_id, amount) VALUES ($1, $2)

If the driver says argument $2, _extract_insert_column_name(...) maps $2 to amount.

That is why the final user message can become:

Invalid data type for column 'amount': expected int, got str.

instead of a low-level driver message.

Where to edit:

• _describe_argument_type_error(...) to support new driver patterns
• _extract_insert_column_name(...) if you want broader SQL parsing
• _extract_sql_text(...) if your wrapped DB errors store SQL in a different place

Enum parsing

_describe_enum_error(...) handles raw messages like:

• invalid input value for enum status_enum: "donee"

It turns them into cleaner guidance.

Where to edit:

• _describe_enum_error(...)

Constraint errors

Constraint messages are handled inside _database_message(...) and _database_status(...).

Current mapping:

• duplicate key -> status 409, code duplicate_resource
• foreign key -> status 422, code invalid_reference
• not-null -> status 422, code missing_required_field
• check constraint -> status 422, code constraint_violation

This split is important:

• _database_message(...) decides the text shown to the user
• _database_status(...) decides the HTTP code and machine-readable error code

If you change one without the other, you can create inconsistent behavior.

Database unavailable errors

Unavailable DB errors currently map to 503, except TimeoutError, which maps to 504.

Where to edit:

• DB_UNAVAILABLE_EXCEPTION_TYPES
• _database_message(...)
• _database_status(...)
• the explicit TimeoutError branch in describe(...)

Programming DB errors

Programming errors are server-side query problems, not user-input problems.

Examples:

• undefined table
• undefined column
• SQL syntax error

These currently return a safe message like:

• The query referenced a table that does not exist.
• The query referenced a column that does not exist.
• SQL syntax error: please verify your query structure.

And they map to 500 with code database_programming_error.

This is correct because these are developer issues, not client data issues.

5. Fallback internal errors

If no branch matches, the library returns:

• status 500
• code internal_error
• message An unexpected error occurred.

This is the last safety net.

Only change this if you want a different global default.

How To Change Exception Responses Safely

Change only the message

Edit one of these:

• _database_message(...)
• _safe_client_message(...)
• _normalize_validation_errors(...)
• the relevant branch in ErrorHandler.describe(...)

Use this when you want the same status code and same response shape, but different user text.

Change only the status code

Edit one of these:

• _database_status(...)
• the relevant branch in ErrorHandler.describe(...)

Use this when the payload shape is correct but the HTTP semantics are wrong.

Change only the JSON payload shape

Edit one of these:

• ErrorHandler.build_payload(...)
• create_response(...) in responses.py

Use this when frontend or consumers need extra keys like request_id, trace_id, a standard meta object, or a different nested error structure.

Change response formatting for normal non-exception routes

Edit create_response(...) in responses.py.

That function controls:

• success payloads
• pagination payloads
• error payload aliases
• optional schema validation for response data

Important distinction:

• create_response(...) is for normal route returns
• unified_exception_handler(...) is for exceptions

They are related, but they are not the same code path.

responses.py Deep Knowledge

value_correction(...)

This is a recursive normalizer.

If you want to support more output types, add them here.

Safe future edits:

• UUID -> string
• Path -> string
• custom dataclass -> dictionary

create_response(...)

This function has five main jobs:

• build success responses
• build direct error responses
• validate response data against a schema when provided
• validate the core pagination structure when pagination is present
• attach a meta object to every JSON response body

Deep flow:

1. set success from the HTTP status range
2. set response_code
3. normalize meta into a dictionary
4. serialize data through value_correction(...)
5. if schema exists, validate before storing data
6. if response-schema validation fails, convert it into a 422 payload
7. if pagination exists, validate page, rows, and total_rows
8. if pagination validation fails, convert it into a 422 payload
9. if error_message exists or status is >= 400, build the error object
10. return JSONResponse or raw dictionary depending on as_json_response

Where to edit:

• _serialize_meta(...) for response metadata rules
• value_correction(...) for data normalization
• _format_validation_errors(...) for response-schema validation detail formatting
• _serialize_pagination(...) for pagination rules
• _set_error_aliases(...) if you want to change legacy keys
• create_response(...) itself for top-level shape or new optional keys

logging.py Deep Knowledge

logging.py is intentionally small.

get_logger(...):

• ensures one stream handler exists
• optionally creates a file handler
• reuses file handlers through _FILE_HANDLERS so repeated calls do not duplicate log lines

logs(...):

• converts text level like "error" into a real logging level
• delegates to a logger returned by get_logger(...)

If you see duplicate log lines in future, check _FILE_HANDLERS and handler attachment first.

Safe Editing Rules

When updating the library, keep these rules:

• put specific exception checks before generic ones
• keep validation handling separate from common 400 errors
• keep database message logic in _database_message(...)
• keep database status logic in _database_status(...)
• keep JSON shape logic in build_payload(...) or create_response(...)
• if you add a new public function, also export it from src/aniket_tools/__init__.py

Reading Order For New Contributors

If you want to understand the code quickly, read in this order:

1. src/aniket_tools/__init__.py
2. src/aniket_tools/responses.py
3. src/aniket_tools/logging.py
4. src/aniket_tools/exceptions.py

Inside exceptions.py, read in this order:

1. grouped exception types
2. ErrorInfo
3. ApiError
4. _normalize_validation_errors(...)
5. _database_message(...)
6. _database_status(...)
7. ErrorHandler.describe(...)
8. ErrorHandler.build_payload(...)
9. unified_exception_handler(...)