# 🚀 ADjango
📊 **Coverage 70%**
> Sometimes I use this in different projects, so I decided to put it on pypi
`ADjango` is a comprehensive library that enhances Django development with Django REST Framework (DRF) and Celery
integration. It provides essential tools including
asynchronous `managers`, `services`, `serializers`, `decorators`, `exceptions` and more utilities for `async`
programming, Celery task scheduling, `transaction` management, and much more to streamline your Django DRF Celery
development workflow.
- [Installation 🛠️](#installation-️)
- [Settings ⚙️](#settings-️)
- [Overview](#overview)
- [Manager \& Services 🛎️](#manager--services-️)
- [Utils 🔧](#utils-)
- [Mixins 🎨](#mixins-)
- [Decorators 🎀](#decorators-)
- [Exceptions 🚨](#exceptions-)
- [Serializers 🔧](#serializers-)
- [Management](#management)
- [Celery 🔥](#celery-)
- [Management Commands](#management-commands)
- [@task Decorator](#task-decorator)
- [Tasker - Task Scheduler](#tasker---task-scheduler)
- [Email Sending via Celery](#email-sending-via-celery)
- [Other](#other)
## Installation 🛠️
```bash
pip install adjango
```
## Settings ⚙️
- ### Add the application to the project
```python
INSTALLED_APPS = [
# ...
'adjango',
]
```
- ### In `settings.py` set the params
```python
# settings.py
# NONE OF THE PARAMETERS ARE REQUIRED
# For usage @a/controller decorators
LOGIN_URL = '/login/'
# optional
ADJANGO_BACKENDS_APPS = BASE_DIR / 'apps' # for management commands
ADJANGO_FRONTEND_APPS = BASE_DIR.parent / 'frontend' / 'src' / 'apps' # for management commands
ADJANGO_APPS_PREPATH = 'apps.' # if apps in BASE_DIR/apps/app1,app2...
ADJANGO_UNCAUGHT_EXCEPTION_HANDLING_FUNCTION = ... # Read about @acontroller, @controller
ADJANGO_CONTROLLERS_LOGGER_NAME = 'global' # only for usage @a/controller decorators
ADJANGO_CONTROLLERS_LOGGING = True # only for usage @a/controller decorators
ADJANGO_EMAIL_LOGGER_NAME = 'email' # for send_emails_task logging
```
```python
MIDDLEWARE = [
...
# add request.ip in views if u need
'adjango.middleware.IPAddressMiddleware',
...
]
```
## Overview
Most functions, if available in asynchronous form, are also available in synchronous form.
### Manager & Services 🛎️
A simple example and everything is immediately clear...
```python
from adjango.fields import AManyToManyField
from adjango.managers.base import AManager
from adjango.services.base import ABaseService
from adjango.models import AModel
from adjango.models.base import AAbstractUser
from adjango.models.polymorphic import APolymorphicModel
...
... # Service layer usage
...
# services/user.py
if TYPE_CHECKING:
from apps.core.models import User
class UserService(ABaseService):
def __init__(self, obj: 'User') -> None:
super().__init__(obj)
self.user = obj
def get_full_name(self) -> str:
return f"{self.user.first_name} {self.user.last_name}"
# models/user.py (User redefinition)
class User(AAbstractUser):
...
@property
def service(self) -> UserService:
return UserService(self)
# and u can use with nice type hints:
user = await User.objects.aget(id=1)
full_name = user.service.get_full_name()
...
... # Other best features
...
# models/commerce.py
class Product(APolymorphicModel):
name = CharField(max_length=100)
class Order(AModel):
user = ForeignKey(User, CASCADE)
products = AManyToManyField(Product)
# The following is now possible...
products = await Product.objects.aall()
products = await Product.objects.afilter(name='name')
# Returns an object or None if not found
order = await Order.objects.agetorn(id=69) # aget or none
if not order: raise
# We install products in the order
await order.products.aset(products)
# Or queryset right away...
await order.products.aset(
Product.objects.filter(name='name')
)
await order.products.aadd(products[0])
# We get the order again without associated objects
order: Order = await Order.objects.aget(id=69)
# Retrieve related objects asynchronously.
order.user = await order.arelated('user')
products = await order.products.aall()
# Works the same with intermediate processing/query filters
orders = await Order.objects.prefetch_related('products').aall()
for o in orders:
for p in o.products.all():
print(p.id)
# thk u
```
### Utils 🔧
`aall`, `afilter`, `arelated`, and so on are available as individual functions
```python
from adjango.utils.funcs import (
aall, getorn, agetorn,
afilter, aset, aadd, arelated
)
```
### Mixins 🎨
```python
from adjango.models.mixins import (
ACreatedAtMixin, ACreatedAtIndexedMixin, ACreatedAtEditableMixin,
AUpdatedAtMixin, AUpdatedAtIndexedMixin,
ACreatedUpdatedAtMixin, ACreatedUpdatedAtIndexedMixin
)
class EventProfile(ACreatedUpdatedAtIndexedMixin):
event = ForeignKey('events.Event', CASCADE, 'members', verbose_name=_('Event'))
@property
def service(self) -> EventProfileService:
return EventProfileService(self)
```
### Decorators 🎀
- `aforce_data`
The `aforce_data` decorator combines data from the `GET`, `POST` and `JSON` body
request in `request.data`. This makes it easy to access all request data in one place.
- `aatomic`
An asynchronous decorator that wraps function into a transactional context using `AsyncAtomicContextManager`. If an
exception occurs, all database changes are rolled back.
- `acontroller/controller`
Decorators that provide automatic logging and exception handling for views. The `acontroller` is for async
views, `controller` is for sync views. They do NOT wrap functions in transactions (use `@aatomic` for that).
```python
from adjango.adecorators import acontroller
from adjango.decorators import controller
@acontroller(name='My View', logger='custom_logger', log_name=True, log_time=True)
async def my_view(request):
pass
@acontroller('One More View')
async def my_view_one_more(request):
pass
@controller(name='Sync View', auth_required=True, log_time=True)
def my_sync_view(request):
pass
```
- These decorators automatically catch uncaught exceptions and log them if the logger is configured
via `ADJANGO_CONTROLLERS_LOGGER_NAME` and `ADJANGO_CONTROLLERS_LOGGING`.
- The `controller` decorator also supports authentication checking with `auth_required` parameter.
- You can also implement the interface:
```python
class IHandlerControllerException(ABC):
@staticmethod
@abstractmethod
def handle(fn_name: str, request: WSGIRequest | ASGIRequest, e: Exception, *args, **kwargs) -> None:
"""
An example of an exception handling function.
:param fn_name: The name of the function where the exception occurred.
:param request: The request object (WSGIRequest or ASGIRequest).
:param e: The exception to be handled.
:param args: Positional arguments passed to the function.
:param kwargs: Named arguments passed to the function.
:return: None
"""
pass
```
and use `handle` to get an uncaught exception:
```python
# settings.py
from adjango.handlers import HCE # use my example if u need
ADJANGO_UNCAUGHT_EXCEPTION_HANDLING_FUNCTION = HCE.handle
```
### Exceptions 🚨
`ADjango` provides convenient classes for generating API exceptions with proper HTTP statuses and structured error
messages.
```python
from adjango.exceptions.base import (
ApiExceptionGenerator,
ModelApiExceptionGenerator,
ModelApiExceptionBaseVariant as MAEBV
)
# General API exceptions
raise ApiExceptionGenerator('Специальная ошибка', 500)
raise ApiExceptionGenerator('Специальная ошибка', 500, 'special_error')
raise ApiExceptionGenerator(
'Неверные данные',
400,
extra={'field': 'email'}
)
# Model exceptions
from apps.commerce.models import Order
raise ModelApiExceptionGenerator(Order, MAEBV.DoesNotExist)
raise ModelApiExceptionGenerator(
Order
MAEBV.AlreadyExists,
code = "order_exists",
extra = {"id": 123}
)
# Available exception variants for models:
# DoesNotExist, AlreadyExists, InvalidData, AccessDenied,
# NotAcceptable, Expired, InternalServerError, AlreadyUsed,
# NotUsed, NotAvailable, TemporarilyUnavailable,
# ConflictDetected, LimitExceeded, DependencyMissing, Deprecated
```
### Serializers 🔧
`ADjango` extends `Django REST Framework` serializers to support asynchronous
operations, making it easier to handle data in async views.
Support methods like `adata`, `avalid_data`, `ais_valid`, and `asave`.
```python
from adjango.querysets.base import AQuerySet
from adjango.aserializers import (
AModelSerializer, ASerializer, AListSerializer
)
from adjango.serializers import dynamic_serializer
...
class ConsultationPublicSerializer(AModelSerializer):
clients = UserPublicSerializer(many=True, read_only=True)
psychologists = UserPsyPublicSerializer(many=True, read_only=True)
config = ConsultationConfigSerializer(read_only=True)
class Meta:
model = Consultation
fields = '__all__'
# From the complete serializer we cut off the pieces into smaller ones
ConsultationSerializerTier1 = dynamic_serializer(
ConsultationPublicSerializer, ('id', 'date',)
)
ConsultationSerializerTier2 = dynamic_serializer(
ConsultationPublicSerializer, (
'id', 'date', 'psychologists', 'clients', 'config'
), {
'psychologists': UserPublicSerializer(many=True), # overridden
}
)
# Use it, in compact format
@acontroller('Completed Consultations')
@api_view(('GET',))
@permission_classes((IsAuthenticated,))
async def consultations_completed(request):
page = int(request.query_params.get('page', 1))
page_size = int(request.query_params.get('page_size', 10))
return Response({
'results': await ConsultationSerializerTier2(
await request.user.completed_consultations[
(page - 1) * page_size:page * page_size
].aall(),
many=True,
context={'request': request}
).adata
}, status=200)
...
class UserService(ABaseService['User']):
...
@property
def completed_consultations(self) -> AQuerySet['Consultation']:
"""
Returns an optimized AQuerySet of all completed consultations of the user
(both psychologist and client).
"""
from apps.psychology.models import Consultation
now_ = now()
return Consultation.objects.defer(
'communication_type',
'language',
'reserved_by',
'notifies',
'cancel_initiator',
'original_consultation',
'consultations_feedbacks',
).select_related(
'config',
'conference',
).prefetch_related(
'clients',
'psychologists',
).filter(
Q(
Q(clients=self.user) | Q(psychologists=self.user),
status=Consultation.Status.PAID,
date__isnull=False,
date__lt=now_,
consultations_feedbacks__user=self.user,
) |
Q(
Q(clients=self) | Q(psychologists=self.user),
status=Consultation.Status.CANCELLED,
date__isnull=False,
)
).distinct().order_by('-updated_at')
...
```
### Management
- `copy_project`
Documentation in the _py_ module itself - **[copy_project](adjango/management/commands/copy_project.py)**
ADjango ships with extra management commands to speed up project scaffolding.
- `astartproject` — clones the [adjango-template](https://github.com/Artasov/adjango-template)
into the given directory and strips its Git history.
```bash
django-admin astartproject myproject
```
- `astartup` — creates an app skeleton inside `apps/` and registers it in
`INSTALLED_APPS`.
```bash
python manage.py astartup blog
```
After running the command you will have the following structure:
```sh
apps/
blog/
controllers/base.py
models/base.py
services/base.py
serializers/base.py
tests/base.py
```
- `newentities` — generates empty exception, model, service, serializer and
test stubs for the specified models in the target app.
```bash
python manage.py newentities order apps.commerce Order,Product,Price
```
Or create a single model:
```bash
python manage.py newentities order apps.commerce Order
```
### Celery 🔥
ADjango provides convenient tools for working with Celery: management commands, decorators, and task scheduler.
For Celery configuration in Django, refer to
the [official Celery documentation](https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html).
#### Management Commands
- `celeryworker` — starts Celery Worker with default settings
```bash
python manage.py celeryworker
python manage.py celeryworker --pool=solo --loglevel=info -E
python manage.py celeryworker --concurrency=4 --queues=high_priority,default
```
- `celerybeat` — starts Celery Beat scheduler for periodic tasks
```bash
python manage.py celerybeat
python manage.py celerybeat --loglevel=debug
```
- `celerypurge` — clears Celery queues from unfinished tasks
```bash
python manage.py celerypurge # clear all queues
python manage.py celerypurge --queue=high # clear specific queue
```
#### @task Decorator
The `@task` decorator automatically logs Celery task execution, including errors:
```python
from celery import shared_task
from adjango.decorators import task
@shared_task
@task(logger="global")
def my_background_task(param1: str, param2: int) -> bool:
"""
Task with automatic execution logging.
"""
# your code here
return True
```
**What the decorator provides:**
- ✅ Automatic logging of task start and completion
- ✅ Logging of task parameters
- ✅ Detailed error logging with stack trace
- ✅ Flexible logger configuration for different tasks
#### Tasker - Task Scheduler
The `Tasker` class provides convenient methods for scheduling and managing Celery tasks:
```python
from adjango.utils.celery.tasker import Tasker
# Immediate execution
task_id = Tasker.put(task=my_task, param1='value')
# Delayed execution (in 60 seconds)
task_id = Tasker.put(task=my_task, countdown=60, param1='value')
# Execution at specific time
from datetime import datetime
task_id = Tasker.put(
task=my_task,
eta=datetime(2024, 12, 31, 23, 59),
param1='value'
)
# Cancel task by ID
Tasker.cancel_task(task_id)
# One-time task via Celery Beat (sync)
Tasker.beat(
task=my_task,
name='one_time_task',
schedule_time=datetime(2024, 10, 10, 14, 30),
param1='value'
)
# Periodic task via Celery Beat (sync)
Tasker.beat(
task=my_task,
name='hourly_cleanup',
interval=3600, # every hour in seconds
param1='value'
)
# Crontab schedule via Celery Beat (sync)
Tasker.beat(
task=my_task,
name='daily_report',
crontab={'hour': 7, 'minute': 30}, # every day at 7:30 AM
param1='value'
)
# Async version of beat is also available
await Tasker.abeat(
task=my_task,
name='async_task',
interval=1800, # every 30 minutes
param1='value'
)
```
#### Email Sending via Celery
ADjango includes a ready-to-use task for sending emails with templates:
```python
from adjango.tasks import send_emails_task
from adjango.utils.mail import send_emails
# Synchronous sending
send_emails(
subject='Welcome!',
emails=('user@example.com',),
template='emails/welcome.html',
context={'user': 'John Doe'}
)
# Asynchronous sending via Celery
send_emails_task.delay(
subject='Hello!',
emails=('user@example.com',),
template='emails/hello.html',
context={'message': 'Welcome to our service!'}
)
# Via Tasker with delayed execution
Tasker.put(
task=send_emails_task,
subject='Reminder',
emails=('user@example.com',),
template='emails/reminder.html',
context={'deadline': '2024-12-31'},
countdown=3600 # send in an hour
)
```
### Other
- `AsyncAtomicContextManager`🧘
An asynchronous context manager for working with transactions, which ensures the atomicity of operations.
```python
from adjango.utils.base import AsyncAtomicContextManager
async def some_function():
async with AsyncAtomicContextManager():
...
```
Raw data
{
"_id": null,
"home_page": "https://github.com/Artasov/adjango",
"name": "adjango",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "adjango django utils funcs features async managers services",
"author": "xlartas",
"author_email": "ivanhvalevskey@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/96/5e/b157dd6229c16324788c101a675f4674571e9c1e2e79d660d0122e55e748/adjango-0.8.0.tar.gz",
"platform": null,
"description": "# \ud83d\ude80 ADjango\r\n\r\n\ud83d\udcca **Coverage 70%**\r\n\r\n> Sometimes I use this in different projects, so I decided to put it on pypi\r\n\r\n`ADjango` is a comprehensive library that enhances Django development with Django REST Framework (DRF) and Celery\r\nintegration. It provides essential tools including\r\nasynchronous `managers`, `services`, `serializers`, `decorators`, `exceptions` and more utilities for `async`\r\nprogramming, Celery task scheduling, `transaction` management, and much more to streamline your Django DRF Celery\r\ndevelopment workflow.\r\n\r\n- [Installation \ud83d\udee0\ufe0f](#installation-\ufe0f)\r\n- [Settings \u2699\ufe0f](#settings-\ufe0f)\r\n- [Overview](#overview)\r\n - [Manager \\& Services \ud83d\udece\ufe0f](#manager--services-\ufe0f)\r\n - [Utils \ud83d\udd27](#utils-)\r\n - [Mixins \ud83c\udfa8](#mixins-)\r\n - [Decorators \ud83c\udf80](#decorators-)\r\n - [Exceptions \ud83d\udea8](#exceptions-)\r\n - [Serializers \ud83d\udd27](#serializers-)\r\n - [Management](#management)\r\n - [Celery \ud83d\udd25](#celery-)\r\n - [Management Commands](#management-commands)\r\n - [@task Decorator](#task-decorator)\r\n - [Tasker - Task Scheduler](#tasker---task-scheduler)\r\n - [Email Sending via Celery](#email-sending-via-celery)\r\n - [Other](#other)\r\n\r\n## Installation \ud83d\udee0\ufe0f\r\n\r\n```bash\r\npip install adjango\r\n```\r\n\r\n## Settings \u2699\ufe0f\r\n\r\n- ### Add the application to the project\r\n\r\n ```python\r\n INSTALLED_APPS = [\r\n # ...\r\n 'adjango',\r\n ]\r\n ```\r\n\r\n- ### In `settings.py` set the params\r\n\r\n ```python\r\n # settings.py\r\n \r\n # NONE OF THE PARAMETERS ARE REQUIRED \r\n \r\n # For usage @a/controller decorators\r\n LOGIN_URL = '/login/' \r\n \r\n # optional\r\n ADJANGO_BACKENDS_APPS = BASE_DIR / 'apps' # for management commands\r\n ADJANGO_FRONTEND_APPS = BASE_DIR.parent / 'frontend' / 'src' / 'apps' # for management commands\r\n ADJANGO_APPS_PREPATH = 'apps.' # if apps in BASE_DIR/apps/app1,app2...\r\n ADJANGO_UNCAUGHT_EXCEPTION_HANDLING_FUNCTION = ... # Read about @acontroller, @controller\r\n ADJANGO_CONTROLLERS_LOGGER_NAME = 'global' # only for usage @a/controller decorators\r\n ADJANGO_CONTROLLERS_LOGGING = True # only for usage @a/controller decorators\r\n ADJANGO_EMAIL_LOGGER_NAME = 'email' # for send_emails_task logging\r\n ```\r\n\r\n ```python\r\n MIDDLEWARE = [\r\n ...\r\n # add request.ip in views if u need\r\n 'adjango.middleware.IPAddressMiddleware', \r\n ...\r\n ]\r\n ```\r\n\r\n## Overview\r\n\r\nMost functions, if available in asynchronous form, are also available in synchronous form.\r\n\r\n### Manager & Services \ud83d\udece\ufe0f\r\n\r\nA simple example and everything is immediately clear...\r\n\r\n```python\r\nfrom adjango.fields import AManyToManyField\r\nfrom adjango.managers.base import AManager\r\nfrom adjango.services.base import ABaseService\r\nfrom adjango.models import AModel\r\nfrom adjango.models.base import AAbstractUser\r\nfrom adjango.models.polymorphic import APolymorphicModel\r\n\r\n...\r\n... # Service layer usage\r\n...\r\n\r\n# services/user.py\r\nif TYPE_CHECKING:\r\n from apps.core.models import User\r\n\r\n\r\nclass UserService(ABaseService):\r\n def __init__(self, obj: 'User') -> None:\r\n super().__init__(obj)\r\n self.user = obj\r\n\r\n def get_full_name(self) -> str:\r\n return f\"{self.user.first_name} {self.user.last_name}\"\r\n\r\n\r\n# models/user.py (User redefinition)\r\nclass User(AAbstractUser):\r\n ...\r\n\r\n @property\r\n def service(self) -> UserService:\r\n return UserService(self)\r\n\r\n\r\n# and u can use with nice type hints:\r\nuser = await User.objects.aget(id=1)\r\nfull_name = user.service.get_full_name()\r\n\r\n...\r\n... # Other best features\r\n...\r\n\r\n\r\n# models/commerce.py\r\nclass Product(APolymorphicModel):\r\n name = CharField(max_length=100)\r\n\r\n\r\nclass Order(AModel):\r\n user = ForeignKey(User, CASCADE)\r\n products = AManyToManyField(Product)\r\n\r\n\r\n# The following is now possible...\r\nproducts = await Product.objects.aall()\r\nproducts = await Product.objects.afilter(name='name')\r\n# Returns an object or None if not found\r\norder = await Order.objects.agetorn(id=69) # aget or none\r\nif not order: raise\r\n\r\n# We install products in the order\r\nawait order.products.aset(products)\r\n# Or queryset right away...\r\nawait order.products.aset(\r\n Product.objects.filter(name='name')\r\n)\r\nawait order.products.aadd(products[0])\r\n\r\n# We get the order again without associated objects\r\norder: Order = await Order.objects.aget(id=69)\r\n# Retrieve related objects asynchronously.\r\norder.user = await order.arelated('user')\r\nproducts = await order.products.aall()\r\n# Works the same with intermediate processing/query filters\r\norders = await Order.objects.prefetch_related('products').aall()\r\nfor o in orders:\r\n for p in o.products.all():\r\n print(p.id)\r\n# thk u\r\n```\r\n\r\n### Utils \ud83d\udd27\r\n\r\n`aall`, `afilter`, `arelated`, and so on are available as individual functions\r\n\r\n ```python\r\n from adjango.utils.funcs import (\r\n aall, getorn, agetorn,\r\n afilter, aset, aadd, arelated\r\n)\r\n ```\r\n\r\n### Mixins \ud83c\udfa8\r\n\r\n```python\r\nfrom adjango.models.mixins import (\r\n ACreatedAtMixin, ACreatedAtIndexedMixin, ACreatedAtEditableMixin,\r\n AUpdatedAtMixin, AUpdatedAtIndexedMixin,\r\n ACreatedUpdatedAtMixin, ACreatedUpdatedAtIndexedMixin\r\n)\r\n\r\n\r\nclass EventProfile(ACreatedUpdatedAtIndexedMixin):\r\n event = ForeignKey('events.Event', CASCADE, 'members', verbose_name=_('Event'))\r\n\r\n @property\r\n def service(self) -> EventProfileService:\r\n return EventProfileService(self)\r\n```\r\n\r\n### Decorators \ud83c\udf80\r\n\r\n- `aforce_data`\r\n\r\n The `aforce_data` decorator combines data from the `GET`, `POST` and `JSON` body\r\n request in `request.data`. This makes it easy to access all request data in one place.\r\n\r\n- `aatomic`\r\n\r\n An asynchronous decorator that wraps function into a transactional context using `AsyncAtomicContextManager`. If an\r\n exception occurs, all database changes are rolled back.\r\n\r\n- `acontroller/controller`\r\n\r\n Decorators that provide automatic logging and exception handling for views. The `acontroller` is for async\r\n views, `controller` is for sync views. They do NOT wrap functions in transactions (use `@aatomic` for that).\r\n\r\n ```python\r\n from adjango.adecorators import acontroller\r\n from adjango.decorators import controller\r\n\r\n @acontroller(name='My View', logger='custom_logger', log_name=True, log_time=True)\r\n async def my_view(request):\r\n pass\r\n \r\n @acontroller('One More View')\r\n async def my_view_one_more(request):\r\n pass\r\n\r\n @controller(name='Sync View', auth_required=True, log_time=True)\r\n def my_sync_view(request):\r\n pass\r\n ```\r\n\r\n - These decorators automatically catch uncaught exceptions and log them if the logger is configured\r\n via `ADJANGO_CONTROLLERS_LOGGER_NAME` and `ADJANGO_CONTROLLERS_LOGGING`.\r\n - The `controller` decorator also supports authentication checking with `auth_required` parameter.\r\n - You can also implement the interface:\r\n\r\n ```python\r\n class IHandlerControllerException(ABC):\r\n @staticmethod\r\n @abstractmethod\r\n def handle(fn_name: str, request: WSGIRequest | ASGIRequest, e: Exception, *args, **kwargs) -> None:\r\n \"\"\"\r\n An example of an exception handling function.\r\n \r\n :param fn_name: The name of the function where the exception occurred.\r\n :param request: The request object (WSGIRequest or ASGIRequest).\r\n :param e: The exception to be handled.\r\n :param args: Positional arguments passed to the function.\r\n :param kwargs: Named arguments passed to the function.\r\n \r\n :return: None\r\n \"\"\"\r\n pass\r\n ```\r\n\r\n and use `handle` to get an uncaught exception:\r\n\r\n ```python\r\n # settings.py\r\n from adjango.handlers import HCE # use my example if u need\r\n ADJANGO_UNCAUGHT_EXCEPTION_HANDLING_FUNCTION = HCE.handle\r\n ```\r\n\r\n### Exceptions \ud83d\udea8\r\n\r\n`ADjango` provides convenient classes for generating API exceptions with proper HTTP statuses and structured error\r\nmessages.\r\n\r\n```python\r\nfrom adjango.exceptions.base import (\r\n ApiExceptionGenerator,\r\n ModelApiExceptionGenerator,\r\n ModelApiExceptionBaseVariant as MAEBV\r\n)\r\n\r\n# General API exceptions\r\nraise ApiExceptionGenerator('\u0421\u043f\u0435\u0446\u0438\u0430\u043b\u044c\u043d\u0430\u044f \u043e\u0448\u0438\u0431\u043a\u0430', 500)\r\nraise ApiExceptionGenerator('\u0421\u043f\u0435\u0446\u0438\u0430\u043b\u044c\u043d\u0430\u044f \u043e\u0448\u0438\u0431\u043a\u0430', 500, 'special_error')\r\nraise ApiExceptionGenerator(\r\n '\u041d\u0435\u0432\u0435\u0440\u043d\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435',\r\n 400,\r\n extra={'field': 'email'}\r\n)\r\n\r\n# Model exceptions\r\nfrom apps.commerce.models import Order\r\n\r\nraise ModelApiExceptionGenerator(Order, MAEBV.DoesNotExist)\r\nraise ModelApiExceptionGenerator(\r\n Order\r\nMAEBV.AlreadyExists,\r\ncode = \"order_exists\",\r\nextra = {\"id\": 123}\r\n)\r\n\r\n# Available exception variants for models:\r\n# DoesNotExist, AlreadyExists, InvalidData, AccessDenied,\r\n# NotAcceptable, Expired, InternalServerError, AlreadyUsed,\r\n# NotUsed, NotAvailable, TemporarilyUnavailable, \r\n# ConflictDetected, LimitExceeded, DependencyMissing, Deprecated\r\n```\r\n\r\n### Serializers \ud83d\udd27\r\n\r\n`ADjango` extends `Django REST Framework` serializers to support asynchronous\r\noperations, making it easier to handle data in async views.\r\nSupport methods like `adata`, `avalid_data`, `ais_valid`, and `asave`.\r\n\r\n```python\r\nfrom adjango.querysets.base import AQuerySet\r\nfrom adjango.aserializers import (\r\n AModelSerializer, ASerializer, AListSerializer\r\n)\r\nfrom adjango.serializers import dynamic_serializer\r\n\r\n...\r\n\r\n\r\nclass ConsultationPublicSerializer(AModelSerializer):\r\n clients = UserPublicSerializer(many=True, read_only=True)\r\n psychologists = UserPsyPublicSerializer(many=True, read_only=True)\r\n config = ConsultationConfigSerializer(read_only=True)\r\n\r\n class Meta:\r\n model = Consultation\r\n fields = '__all__'\r\n\r\n\r\n# From the complete serializer we cut off the pieces into smaller ones\r\nConsultationSerializerTier1 = dynamic_serializer(\r\n ConsultationPublicSerializer, ('id', 'date',)\r\n)\r\nConsultationSerializerTier2 = dynamic_serializer(\r\n ConsultationPublicSerializer, (\r\n 'id', 'date', 'psychologists', 'clients', 'config'\r\n ), {\r\n 'psychologists': UserPublicSerializer(many=True), # overridden\r\n }\r\n)\r\n\r\n\r\n# Use it, in compact format\r\n@acontroller('Completed Consultations')\r\n@api_view(('GET',))\r\n@permission_classes((IsAuthenticated,))\r\nasync def consultations_completed(request):\r\n page = int(request.query_params.get('page', 1))\r\n page_size = int(request.query_params.get('page_size', 10))\r\n return Response({\r\n 'results': await ConsultationSerializerTier2(\r\n await request.user.completed_consultations[\r\n (page - 1) * page_size:page * page_size\r\n ].aall(),\r\n many=True,\r\n context={'request': request}\r\n ).adata\r\n }, status=200)\r\n\r\n\r\n...\r\n\r\n\r\nclass UserService(ABaseService['User']):\r\n ...\r\n\r\n @property\r\n def completed_consultations(self) -> AQuerySet['Consultation']:\r\n \"\"\"\r\n Returns an optimized AQuerySet of all completed consultations of the user\r\n (both psychologist and client).\r\n \"\"\"\r\n from apps.psychology.models import Consultation\r\n now_ = now()\r\n return Consultation.objects.defer(\r\n 'communication_type',\r\n 'language',\r\n 'reserved_by',\r\n 'notifies',\r\n 'cancel_initiator',\r\n 'original_consultation',\r\n 'consultations_feedbacks',\r\n ).select_related(\r\n 'config',\r\n 'conference',\r\n ).prefetch_related(\r\n 'clients',\r\n 'psychologists',\r\n ).filter(\r\n Q(\r\n Q(clients=self.user) | Q(psychologists=self.user),\r\n status=Consultation.Status.PAID,\r\n date__isnull=False,\r\n date__lt=now_,\r\n consultations_feedbacks__user=self.user,\r\n ) |\r\n Q(\r\n Q(clients=self) | Q(psychologists=self.user),\r\n status=Consultation.Status.CANCELLED,\r\n date__isnull=False,\r\n )\r\n ).distinct().order_by('-updated_at')\r\n\r\n ...\r\n```\r\n\r\n### Management\r\n\r\n- `copy_project`\r\n Documentation in the _py_ module itself - **[copy_project](adjango/management/commands/copy_project.py)**\r\n\r\nADjango ships with extra management commands to speed up project scaffolding.\r\n\r\n- `astartproject` \u2014 clones the [adjango-template](https://github.com/Artasov/adjango-template)\r\n into the given directory and strips its Git history.\r\n\r\n ```bash\r\n django-admin astartproject myproject\r\n ```\r\n\r\n- `astartup` \u2014 creates an app skeleton inside `apps/` and registers it in\r\n `INSTALLED_APPS`.\r\n\r\n ```bash\r\n python manage.py astartup blog\r\n ```\r\n\r\n After running the command you will have the following structure:\r\n\r\n ```sh\r\n apps/\r\n blog/\r\n controllers/base.py\r\n models/base.py\r\n services/base.py\r\n serializers/base.py\r\n tests/base.py\r\n ```\r\n\r\n- `newentities` \u2014 generates empty exception, model, service, serializer and\r\n test stubs for the specified models in the target app.\r\n\r\n ```bash\r\n python manage.py newentities order apps.commerce Order,Product,Price\r\n ```\r\n\r\n Or create a single model:\r\n\r\n ```bash\r\n python manage.py newentities order apps.commerce Order\r\n ```\r\n\r\n### Celery \ud83d\udd25\r\n\r\nADjango provides convenient tools for working with Celery: management commands, decorators, and task scheduler.\r\n\r\nFor Celery configuration in Django, refer to\r\nthe [official Celery documentation](https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html).\r\n\r\n#### Management Commands\r\n\r\n- `celeryworker` \u2014 starts Celery Worker with default settings\r\n\r\n ```bash\r\n python manage.py celeryworker\r\n python manage.py celeryworker --pool=solo --loglevel=info -E\r\n python manage.py celeryworker --concurrency=4 --queues=high_priority,default\r\n ```\r\n\r\n- `celerybeat` \u2014 starts Celery Beat scheduler for periodic tasks\r\n\r\n ```bash\r\n python manage.py celerybeat\r\n python manage.py celerybeat --loglevel=debug\r\n ```\r\n\r\n- `celerypurge` \u2014 clears Celery queues from unfinished tasks\r\n\r\n ```bash\r\n python manage.py celerypurge # clear all queues\r\n python manage.py celerypurge --queue=high # clear specific queue\r\n ```\r\n\r\n#### @task Decorator\r\n\r\nThe `@task` decorator automatically logs Celery task execution, including errors:\r\n\r\n```python\r\nfrom celery import shared_task\r\nfrom adjango.decorators import task\r\n\r\n\r\n@shared_task\r\n@task(logger=\"global\")\r\ndef my_background_task(param1: str, param2: int) -> bool:\r\n \"\"\"\r\n Task with automatic execution logging.\r\n \"\"\"\r\n # your code here\r\n return True\r\n```\r\n\r\n**What the decorator provides:**\r\n\r\n- \u2705 Automatic logging of task start and completion\r\n- \u2705 Logging of task parameters\r\n- \u2705 Detailed error logging with stack trace\r\n- \u2705 Flexible logger configuration for different tasks\r\n\r\n#### Tasker - Task Scheduler\r\n\r\nThe `Tasker` class provides convenient methods for scheduling and managing Celery tasks:\r\n\r\n```python\r\nfrom adjango.utils.celery.tasker import Tasker\r\n\r\n# Immediate execution\r\ntask_id = Tasker.put(task=my_task, param1='value')\r\n\r\n# Delayed execution (in 60 seconds)\r\ntask_id = Tasker.put(task=my_task, countdown=60, param1='value')\r\n\r\n# Execution at specific time\r\nfrom datetime import datetime\r\n\r\ntask_id = Tasker.put(\r\n task=my_task,\r\n eta=datetime(2024, 12, 31, 23, 59),\r\n param1='value'\r\n)\r\n\r\n# Cancel task by ID\r\nTasker.cancel_task(task_id)\r\n\r\n# One-time task via Celery Beat (sync)\r\nTasker.beat(\r\n task=my_task,\r\n name='one_time_task',\r\n schedule_time=datetime(2024, 10, 10, 14, 30),\r\n param1='value'\r\n)\r\n\r\n# Periodic task via Celery Beat (sync)\r\nTasker.beat(\r\n task=my_task,\r\n name='hourly_cleanup',\r\n interval=3600, # every hour in seconds\r\n param1='value'\r\n)\r\n\r\n# Crontab schedule via Celery Beat (sync)\r\nTasker.beat(\r\n task=my_task,\r\n name='daily_report',\r\n crontab={'hour': 7, 'minute': 30}, # every day at 7:30 AM\r\n param1='value'\r\n)\r\n\r\n# Async version of beat is also available\r\nawait Tasker.abeat(\r\n task=my_task,\r\n name='async_task',\r\n interval=1800, # every 30 minutes\r\n param1='value'\r\n)\r\n```\r\n\r\n#### Email Sending via Celery\r\n\r\nADjango includes a ready-to-use task for sending emails with templates:\r\n\r\n```python\r\nfrom adjango.tasks import send_emails_task\r\nfrom adjango.utils.mail import send_emails\r\n\r\n# Synchronous sending\r\nsend_emails(\r\n subject='Welcome!',\r\n emails=('user@example.com',),\r\n template='emails/welcome.html',\r\n context={'user': 'John Doe'}\r\n)\r\n\r\n# Asynchronous sending via Celery\r\nsend_emails_task.delay(\r\n subject='Hello!',\r\n emails=('user@example.com',),\r\n template='emails/hello.html',\r\n context={'message': 'Welcome to our service!'}\r\n)\r\n\r\n# Via Tasker with delayed execution\r\nTasker.put(\r\n task=send_emails_task,\r\n subject='Reminder',\r\n emails=('user@example.com',),\r\n template='emails/reminder.html',\r\n context={'deadline': '2024-12-31'},\r\n countdown=3600 # send in an hour\r\n)\r\n```\r\n\r\n### Other\r\n\r\n- `AsyncAtomicContextManager`\ud83e\uddd8\r\n\r\n An asynchronous context manager for working with transactions, which ensures the atomicity of operations.\r\n\r\n ```python\r\n from adjango.utils.base import AsyncAtomicContextManager\r\n \r\n async def some_function():\r\n async with AsyncAtomicContextManager():\r\n ... \r\n ```\r\n",
"bugtrack_url": null,
"license": null,
"summary": "A library with many features for interacting with Django",
"version": "0.8.0",
"project_urls": {
"Homepage": "https://github.com/Artasov/adjango",
"Source": "https://github.com/Artasov/adjango",
"Tracker": "https://github.com/Artasov/adjango/issues"
},
"split_keywords": [
"adjango",
"django",
"utils",
"funcs",
"features",
"async",
"managers",
"services"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "6db9cda9bbfe9253fabd769318f27c253768a9afd524efea2200081730e4bf34",
"md5": "3af90d6aaf15bf89ec98269a9f98afa2",
"sha256": "23a6c0879b583c807e83df7e1fadea7571b87f30b713c4e07541a2affe2c05ab"
},
"downloads": -1,
"filename": "adjango-0.8.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3af90d6aaf15bf89ec98269a9f98afa2",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 61622,
"upload_time": "2025-08-27T02:05:51",
"upload_time_iso_8601": "2025-08-27T02:05:51.860038Z",
"url": "https://files.pythonhosted.org/packages/6d/b9/cda9bbfe9253fabd769318f27c253768a9afd524efea2200081730e4bf34/adjango-0.8.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "965eb157dd6229c16324788c101a675f4674571e9c1e2e79d660d0122e55e748",
"md5": "5293e05fb793caf42393c7f3ad4507ce",
"sha256": "6dcca0f5111585d31c65b65d51fd672ba69f8a25aa044ca5309be709738e81ca"
},
"downloads": -1,
"filename": "adjango-0.8.0.tar.gz",
"has_sig": false,
"md5_digest": "5293e05fb793caf42393c7f3ad4507ce",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 67996,
"upload_time": "2025-08-27T02:05:53",
"upload_time_iso_8601": "2025-08-27T02:05:53.579669Z",
"url": "https://files.pythonhosted.org/packages/96/5e/b157dd6229c16324788c101a675f4674571e9c1e2e79d660d0122e55e748/adjango-0.8.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-27 02:05:53",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Artasov",
"github_project": "adjango",
"travis_ci": false,
"coveralls": true,
"github_actions": false,
"requirements": [],
"lcname": "adjango"
}
JSON
