Skip to main content

A generic, reusable task management system built with FastAPI and asyncio

Project description

🍆 Brinjal

A generic, reusable task management system built with FastAPI and asyncio. Brinjal provides a foundation for building applications that need to manage long-running tasks with real-time progress updates via Server-Sent Events (SSE).

Features

  • Generic Task Framework: Base Task class that can be extended for any type of work
  • Real-time Updates: Server-Sent Events (SSE) for live progress monitoring
  • Asynchronous Execution: Built on asyncio for non-blocking task processing
  • Web Component: Reusable <task-list> component for displaying tasks
  • Self-contained: No external static file mounting required
  • Easy Integration: Simple router inclusion for any FastAPI application

Quick Start

Installation

# Clone the repository
git clone <your-repo-url>
cd brinjal

# Install dependencies
make install

# Run the development server
make dev

Testing End-to-End

  1. Start the server:

    make dev
    
  2. Create an example task:

    curl -X POST "http://localhost:8000/api/tasks/example_task"
    
  3. Check task status:

    curl "http://localhost:8000/api/tasks/queue"
    
  4. Stream real-time updates:

    # Replace {task_id} with the actual task ID from step 2
    curl "http://localhost:8000/api/tasks/{task_id}/stream"
    
  5. View the web interface:

    • Open http://localhost:8000/api/tasks/test in your browser
    • See the TaskList component in action

Usage in Your Project

Basic Integration

from fastapi import FastAPI
from brinjal.api.router import router

app = FastAPI()
app.include_router(router)

# That's it! Your app now has:
# - /api/tasks/queue - List all tasks
# - /api/tasks/{task_id}/stream - SSE stream for task updates
# - /api/tasks/static/* - Static files (TaskList.js, etc.)

Creating Custom Tasks

from brinjal.task import Task
import time

class MyCustomTask(Task):
    def run(self):
        """Implement your synchronous work here"""
        for i in range(10):
            self.progress = i * 10
            time.sleep(1)
        
        self.status = "done"
        self.progress = 100
        self.heading = "Custom Task Complete"
        self.body = "This task did something amazing!"

Using the TaskList Component

<!DOCTYPE html>
<html>
<head>
    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body>
    <task-list base_url="http://localhost:8000"></task-list>
    <script src="http://localhost:8000/api/tasks/static/TaskList.js"></script>
</body>
</html>

API Reference

Endpoints

  • POST /api/tasks/example_task - Create an example task
  • GET /api/tasks/queue - Get all tasks
  • GET /api/tasks/{task_id}/stream - SSE stream for task updates
  • GET /api/tasks/static/{file} - Static files (TaskList.js, etc.)

Task Model

@dataclass
class Task:
    task_id: str
    status: str  # "pending", "running", "done", "failed"
    progress: int  # 0-100
    img: Optional[str] = None
    heading: Optional[str] = None
    body: Optional[str] = None

Development

# Install dependencies
make install

# Run development server
make dev

# Run tests
make test

# Lint code
make lint

# Format code
make format

# Clean up
make clean

Architecture

  • Task: Base class for all tasks
  • TaskManager: Manages task queue and execution
  • TaskUpdate: Pydantic model for task updates
  • ExampleTask: Sample implementation
  • TaskList.js: Web component for displaying tasks

License

GPL V3

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

brinjal-0.1.0.tar.gz (732.1 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

brinjal-0.1.0-py3-none-any.whl (23.8 kB view details)

Uploaded Python 3

File details

Details for the file brinjal-0.1.0.tar.gz.

File metadata

  • Download URL: brinjal-0.1.0.tar.gz
  • Upload date:
  • Size: 732.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.12.9

File hashes

Hashes for brinjal-0.1.0.tar.gz
Algorithm Hash digest
SHA256 aabb56b0954a60b0f161ec3c35e01a4e94462806b083a192f66b911f3cacd694
MD5 0a81b4f9808d33d48b59cb2e6c618953
BLAKE2b-256 022b66f1cc0267b00425e32569c97a4b7c609c43fe81e46c1ca17388335013c5

See more details on using hashes here.

Provenance

The following attestation bundles were made for brinjal-0.1.0.tar.gz:

Publisher: pypi.yml on sg-s/brinjal

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file brinjal-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: brinjal-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 23.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.12.9

File hashes

Hashes for brinjal-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 23214663713c3ceaf264d7532e6d7e4d81e41bcaa70ac89f20e89474a50eac25
MD5 e8a0ca2f7c29175549d3a5ca3a203c5c
BLAKE2b-256 b71df8cdb1d5964938dc56b67b8c9508ee09ec52e753dc93be795f6be813a7a6

See more details on using hashes here.

Provenance

The following attestation bundles were made for brinjal-0.1.0-py3-none-any.whl:

Publisher: pypi.yml on sg-s/brinjal

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page