# asgi-lifecycle
Flexible ASGI application lifecycle management for Python web frameworks (Django, FastAPI, Starlette, Celery, and more).
## Features
- **Hook-based startup/shutdown**: Register async or sync functions to run on app startup/shutdown.
- **Context-aware**: Hooks receive a rich context object (service type, environment, config, metadata).
- **Service filtering**: Only run hooks for relevant service types (main app, worker, API, scheduler, etc).
- **Singleton manager**: Application-wide lifecycle state and hook registry.
- **Decorator API**: Clean, Pythonic `@lifespan.on_start` and `@lifespan.on_shutdown` decorators.
- **Async and sync support**: Works with both async and sync hooks.
- **Timeouts and error handling**: Robust shutdown with timeouts and logging.
- **Framework-agnostic**: Integrates with any ASGI app, including Django, FastAPI, Starlette, Celery, etc.
## Why asgi-lifespan?
### vs Django's built-in ASGI
- ✅ **Startup/shutdown hooks** vs no lifecycle management
- ✅ **Service type filtering** vs no service differentiation
- ✅ **Priority ordering** vs no execution control
- ✅ **Rich context object** vs no context passing
- ✅ **Error handling & timeouts** vs no error management
- ✅ **Multiple hooks per phase** vs no hook system at all
### vs FastAPI's lifespan
- ✅ Framework-agnostic
- ✅ Multiple hooks per phase
- ✅ Service type filtering
- ✅ Singleton management
## Performance
- **Minimal overhead**: Hooks only run during startup/shutdown
- **Async-first**: Built for modern async Python
- **Timeout protection**: Prevents hanging shutdowns
- **Error isolation**: Failed hooks don't stop others
## Installation
```bash
pip install asgi-lifecycle
```
## Quickstart
```python
from asgi_lifecycle import lifespan, LifespanContext, ServiceType
@lifespan.on_start(priority=1, service_types=["app", "api"])
async def setup_database(context: LifespanContext):
await database.connect()
context.set_metadata("database_connected", True)
@lifespan.on_shutdown(priority=1, service_types=["app", "api"])
async def close_database(context: LifespanContext):
if context.get_metadata("database_connected", False):
await database.disconnect()
```
## API Reference
### Lifespan
- `on_start(priority=0, name=None, service_types=None)`: Decorator to register a startup hook.
- `on_shutdown(priority=0, name=None, service_types=None)`: Decorator to register a shutdown hook.
- `startup(context: LifespanContext)`: Run all startup hooks for the given context.
- `shutdown(context: LifespanContext)`: Run all shutdown hooks for the given context.
- `is_initialized()`: Check if the manager is initialized.
- `reset_instance()`: Reset the singleton (for testing).
### LifespanContext
- `service_type`: The type of service ("app", "worker", "api", "scheduler", "test").
- `environment`: Environment name ("development", "production", etc).
- `config`: Arbitrary config dict.
- `metadata`: Arbitrary metadata dict.
- `get_config(key, default=None)`: Get config value.
- `set_metadata(key, value)`: Set metadata value.
- `get_metadata(key, default=None)`: Get metadata value.
## Example Integrations
### Django ASGI
#### 1. Register Lifecycle Hooks
Add startup and shutdown hooks to your Django settings module. These will run when the ASGI app starts and stops.
```python
# Basic startup hook
@lifespan.on_start()
async def setup_logging(context: LifespanContext):
"""Configure logging when the ASGI app starts."""
logging.basicConfig(level=logging.INFO)
logger.info(f"Starting {context.service_type} in {context.environment}")
# Database setup (only for main app)
@lifespan.on_start(priority=1, service_types=["app"])
async def setup_database(context: LifespanContext):
"""Setup database connections."""
if context.environment == "production":
await database.connect_with_ssl()
else:
await database.connect()
context.set_metadata("database_connected", True)
logger.info("Database connected")
# Shutdown hook
@lifespan.on_shutdown(priority=1, service_types=["app"])
async def close_database(context: LifespanContext):
"""Clean up database connections."""
if context.get_metadata("database_connected", False):
await database.disconnect()
logger.info("Database disconnected")
```
#### 2. Integrate LifespanMiddleware
Wrap your Django ASGI application with `LifespanMiddleware` to enable lifecycle hooks.
```python
# my_django_project/my_django_app/asgi.py
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
import django
django.setup() # Initialize Django before any other imports
from django.core.asgi import get_asgi_application
from asgi_lifecycle import LifespanMiddleware, LifespanContext
# Create the base ASGI application
django_app = get_asgi_application()
# Wrap with lifespan middleware
application = LifespanMiddleware(django_app)
```
#### 4. Import Your Lifecycle Modules
Make sure to import any modules containing lifecycle hooks:
```python
# In your settings.py
import myapp.lifecycle # This registers the hooks
import myapp.database.lifecycle # This too
```
### FastAPI
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from asgi_lifecycle import lifespan, LifespanContext
@asynccontextmanager
async def lifespan_context(app: FastAPI):
# Startup
context = LifespanContext(service_type="api")
await lifespan.startup(context)
yield
# Shutdown
await lifespan.shutdown(context)
app = FastAPI(lifespan=lifespan_context)
```
### Celery Worker
```python
from celery.signals import worker_init, worker_shutdown
from asgiref.sync import async_to_sync
from asgi_lifecycle import lifespan, LifespanContext
@worker_init.connect
def init_worker_process(sender=None, conf=None, **kwargs):
context = LifespanContext(service_type="worker", environment="production", config={"celery_conf": conf})
async_to_sync(lifespan.startup)(context)
@worker_shutdown.connect
def shutdown_worker_process(sender=None, **kwargs):
context = LifespanContext(service_type="worker", environment="production")
async_to_sync(lifespan.shutdown)(context)
```
### Service Types
```python
# Only runs in main app
@lifespan.on_start(service_types=["app"])
async def setup_web_server():
pass
# Only runs in worker
@lifespan.on_start(service_types=["worker"])
async def setup_worker_pool():
pass
# Runs in both
@lifespan.on_start(service_types=["app", "worker"])
async def setup_shared_resources():
pass
```
### Error Handling
```python
@lifespan.on_start(priority=1)
async def setup_database():
try:
await database.connect()
logger.info("Database connected")
except Exception as e:
logger.error(f"Database connection failed: {e}")
# Don't re-raise - let other hooks continue
```
## License
MIT
## Author
Tarek Sanger
## Contributing
Pull requests and issues welcome! See CONTRIBUTING.md.
## Roadmap
- [ ] Auto-discovery of lifecycle modules
- [ ] Built-in health checks
- [ ] Metrics integration
- [ ] Configuration file support
Raw data
{
"_id": null,
"home_page": null,
"name": "asgi-lifecycle",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "asgi, celery, django, fastapi, hooks, lifecycle, lifespan, middleware, shutdown, starlette, startup",
"author": null,
"author_email": "Tarek Sanger <tarek.sanger@me.com>",
"download_url": "https://files.pythonhosted.org/packages/b8/a9/034fb9ffce8130cb72961afeda2af717fae85b94c27871dd2d3135e4a9e1/asgi_lifecycle-0.0.2.tar.gz",
"platform": null,
"description": "# asgi-lifecycle\n\nFlexible ASGI application lifecycle management for Python web frameworks (Django, FastAPI, Starlette, Celery, and more).\n\n## Features\n- **Hook-based startup/shutdown**: Register async or sync functions to run on app startup/shutdown.\n- **Context-aware**: Hooks receive a rich context object (service type, environment, config, metadata).\n- **Service filtering**: Only run hooks for relevant service types (main app, worker, API, scheduler, etc).\n- **Singleton manager**: Application-wide lifecycle state and hook registry.\n- **Decorator API**: Clean, Pythonic `@lifespan.on_start` and `@lifespan.on_shutdown` decorators.\n- **Async and sync support**: Works with both async and sync hooks.\n- **Timeouts and error handling**: Robust shutdown with timeouts and logging.\n- **Framework-agnostic**: Integrates with any ASGI app, including Django, FastAPI, Starlette, Celery, etc.\n\n## Why asgi-lifespan?\n\n### vs Django's built-in ASGI\n- \u2705 **Startup/shutdown hooks** vs no lifecycle management\n- \u2705 **Service type filtering** vs no service differentiation\n- \u2705 **Priority ordering** vs no execution control\n- \u2705 **Rich context object** vs no context passing\n- \u2705 **Error handling & timeouts** vs no error management\n- \u2705 **Multiple hooks per phase** vs no hook system at all\n\n\n### vs FastAPI's lifespan\n- \u2705 Framework-agnostic\n- \u2705 Multiple hooks per phase\n- \u2705 Service type filtering\n- \u2705 Singleton management\n\n\n## Performance\n\n- **Minimal overhead**: Hooks only run during startup/shutdown\n- **Async-first**: Built for modern async Python\n- **Timeout protection**: Prevents hanging shutdowns\n- **Error isolation**: Failed hooks don't stop others\n\n\n## Installation\n\n```bash\npip install asgi-lifecycle\n```\n\n## Quickstart\n\n```python\nfrom asgi_lifecycle import lifespan, LifespanContext, ServiceType\n\n@lifespan.on_start(priority=1, service_types=[\"app\", \"api\"])\nasync def setup_database(context: LifespanContext):\n await database.connect()\n context.set_metadata(\"database_connected\", True)\n\n@lifespan.on_shutdown(priority=1, service_types=[\"app\", \"api\"])\nasync def close_database(context: LifespanContext):\n if context.get_metadata(\"database_connected\", False):\n await database.disconnect()\n\n```\n\n## API Reference\n\n### Lifespan\n- `on_start(priority=0, name=None, service_types=None)`: Decorator to register a startup hook.\n- `on_shutdown(priority=0, name=None, service_types=None)`: Decorator to register a shutdown hook.\n- `startup(context: LifespanContext)`: Run all startup hooks for the given context.\n- `shutdown(context: LifespanContext)`: Run all shutdown hooks for the given context.\n- `is_initialized()`: Check if the manager is initialized.\n- `reset_instance()`: Reset the singleton (for testing).\n\n### LifespanContext\n- `service_type`: The type of service (\"app\", \"worker\", \"api\", \"scheduler\", \"test\").\n- `environment`: Environment name (\"development\", \"production\", etc).\n- `config`: Arbitrary config dict.\n- `metadata`: Arbitrary metadata dict.\n- `get_config(key, default=None)`: Get config value.\n- `set_metadata(key, value)`: Set metadata value.\n- `get_metadata(key, default=None)`: Get metadata value.\n\n## Example Integrations\n\n### Django ASGI\n\n\n#### 1. Register Lifecycle Hooks\n\nAdd startup and shutdown hooks to your Django settings module. These will run when the ASGI app starts and stops.\n\n```python\n# Basic startup hook\n@lifespan.on_start()\nasync def setup_logging(context: LifespanContext):\n \"\"\"Configure logging when the ASGI app starts.\"\"\"\n logging.basicConfig(level=logging.INFO)\n logger.info(f\"Starting {context.service_type} in {context.environment}\")\n\n# Database setup (only for main app)\n@lifespan.on_start(priority=1, service_types=[\"app\"])\nasync def setup_database(context: LifespanContext):\n \"\"\"Setup database connections.\"\"\"\n if context.environment == \"production\":\n await database.connect_with_ssl()\n else:\n await database.connect()\n \n context.set_metadata(\"database_connected\", True)\n logger.info(\"Database connected\")\n\n# Shutdown hook\n@lifespan.on_shutdown(priority=1, service_types=[\"app\"])\nasync def close_database(context: LifespanContext):\n \"\"\"Clean up database connections.\"\"\"\n if context.get_metadata(\"database_connected\", False):\n await database.disconnect()\n logger.info(\"Database disconnected\")\n```\n\n#### 2. Integrate LifespanMiddleware\n\nWrap your Django ASGI application with `LifespanMiddleware` to enable lifecycle hooks.\n\n```python\n# my_django_project/my_django_app/asgi.py\n\nimport os\nos.environ.setdefault(\"DJANGO_SETTINGS_MODULE\", \"myproject.settings\")\n\nimport django\ndjango.setup() # Initialize Django before any other imports\n\nfrom django.core.asgi import get_asgi_application\nfrom asgi_lifecycle import LifespanMiddleware, LifespanContext\n\n# Create the base ASGI application\ndjango_app = get_asgi_application()\n\n# Wrap with lifespan middleware\napplication = LifespanMiddleware(django_app)\n```\n\n#### 4. Import Your Lifecycle Modules\n\nMake sure to import any modules containing lifecycle hooks:\n\n```python\n# In your settings.py\nimport myapp.lifecycle # This registers the hooks\nimport myapp.database.lifecycle # This too\n```\n\n### FastAPI\n```python\nfrom contextlib import asynccontextmanager\nfrom fastapi import FastAPI\nfrom asgi_lifecycle import lifespan, LifespanContext\n\n@asynccontextmanager\nasync def lifespan_context(app: FastAPI):\n # Startup\n context = LifespanContext(service_type=\"api\")\n await lifespan.startup(context)\n \n yield\n \n # Shutdown\n await lifespan.shutdown(context)\n\napp = FastAPI(lifespan=lifespan_context)\n```\n\n### Celery Worker\n```python\nfrom celery.signals import worker_init, worker_shutdown\nfrom asgiref.sync import async_to_sync\nfrom asgi_lifecycle import lifespan, LifespanContext\n\n@worker_init.connect\ndef init_worker_process(sender=None, conf=None, **kwargs):\n context = LifespanContext(service_type=\"worker\", environment=\"production\", config={\"celery_conf\": conf})\n async_to_sync(lifespan.startup)(context)\n\n@worker_shutdown.connect\ndef shutdown_worker_process(sender=None, **kwargs):\n context = LifespanContext(service_type=\"worker\", environment=\"production\")\n async_to_sync(lifespan.shutdown)(context)\n```\n\n### Service Types\n```python\n# Only runs in main app\n@lifespan.on_start(service_types=[\"app\"])\nasync def setup_web_server():\n pass\n\n# Only runs in worker\n@lifespan.on_start(service_types=[\"worker\"])\nasync def setup_worker_pool():\n pass\n\n# Runs in both\n@lifespan.on_start(service_types=[\"app\", \"worker\"])\nasync def setup_shared_resources():\n pass\n```\n\n### Error Handling\n```python\n@lifespan.on_start(priority=1)\nasync def setup_database():\n try:\n await database.connect()\n logger.info(\"Database connected\")\n except Exception as e:\n logger.error(f\"Database connection failed: {e}\")\n # Don't re-raise - let other hooks continue\n```\n\n## License\nMIT\n\n## Author\nTarek Sanger\n\n## Contributing\nPull requests and issues welcome! See CONTRIBUTING.md.\n\n## Roadmap\n\n- [ ] Auto-discovery of lifecycle modules\n- [ ] Built-in health checks\n- [ ] Metrics integration\n- [ ] Configuration file support\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Flexible ASGI application lifecycle management for Python web frameworks.",
"version": "0.0.2",
"project_urls": null,
"split_keywords": [
"asgi",
" celery",
" django",
" fastapi",
" hooks",
" lifecycle",
" lifespan",
" middleware",
" shutdown",
" starlette",
" startup"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "ca40701b302775a5dba85fe160974da63116c17684bce9e32c4038ba962524e7",
"md5": "92350f73977c6edc7d13e464b886fcc7",
"sha256": "145a3bd80e15fd691ebf315bc497a01b0836d3488b2f36ddaae12f58d06efc7a"
},
"downloads": -1,
"filename": "asgi_lifecycle-0.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "92350f73977c6edc7d13e464b886fcc7",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 6996,
"upload_time": "2025-10-11T22:42:04",
"upload_time_iso_8601": "2025-10-11T22:42:04.480647Z",
"url": "https://files.pythonhosted.org/packages/ca/40/701b302775a5dba85fe160974da63116c17684bce9e32c4038ba962524e7/asgi_lifecycle-0.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "b8a9034fb9ffce8130cb72961afeda2af717fae85b94c27871dd2d3135e4a9e1",
"md5": "b4d7b0b603e513500c8b6faa17431721",
"sha256": "f00420095e86b2bb4683c919fa4194bca1ac9d6a16e0a4dc1504a8b83faef70f"
},
"downloads": -1,
"filename": "asgi_lifecycle-0.0.2.tar.gz",
"has_sig": false,
"md5_digest": "b4d7b0b603e513500c8b6faa17431721",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 9062,
"upload_time": "2025-10-11T22:42:05",
"upload_time_iso_8601": "2025-10-11T22:42:05.292344Z",
"url": "https://files.pythonhosted.org/packages/b8/a9/034fb9ffce8130cb72961afeda2af717fae85b94c27871dd2d3135e4a9e1/asgi_lifecycle-0.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-11 22:42:05",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "asgi-lifecycle"
}
JSON