adjango 0.6.6

A library with many features for interacting with Django

🚀 ADjango

Sometimes I use this in different projects, so I decided to put it on pypi

ADjango is a convenient library for simplifying work with Django DRF and other, which offers various useful managers, services, serializers, decorators, utilities for asynchronous programming, a task scheduler for Celery, working with transactions and much more.

• Installation
• Settings
• Overview
  • Manager & Services
  • Utils
  • Mixins
  • Decorators
  • Serializers
  • Management
  • Other

Installation 🛠️

pip install adjango

Settings ⚙️

• Add the application to the project.

  INSTALLED_APPS = [
      # ...
      'adjango',
  ]
  

• In settings.py set the params

  # 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
  

  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...

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['User']):
  def __init__(self, user: 'User') -> None:
    super().__init__(user)
    self.user = user

  def get_full_name(self) -> str:
    return f"{self.user.first_name} {self.user.last_name}"


# models/user.py (User redefinition)
class User(AAbstractUser[UserService]):
  service_class = UserService
  objects = AManager()


# 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):
  # APolymorphicManager() of course here already exists
  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.related('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

from adjango.utils.funcs import (
  aall, getorn, agetorn, 
  afilter, aset, aadd, arelated
)

Mixins 🎨

from adjango.models.mixins import (
    ACreatedAtMixin, ACreatedAtIndexedMixin, ACreatedAtEditableMixin,
    AUpdatedAtMixin, AUpdatedAtIndexedMixin,
    ACreatedUpdatedAtMixin, ACreatedUpdatedAtIndexedMixin
)

class EventProfile(
    ACreatedUpdatedAtIndexedMixin[EventProfileService]
):
    service_class = EventProfileService
    
    event = ForeignKey('events.Event', CASCADE, 'members', verbose_name=_('Event'))

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.

• atomic

  An asynchronous decorator that wraps function into a transactional context. If an exception occurs, all changes are rolled back.

• acontroller/controller

  An asynchronous decorator that wraps function into a transactional context. If an exception occurs, all changes are rolled back.

  from adjango.adecorators import acontroller
  
  @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
  

  • These decorators automatically catch uncaught exceptions and log if the logger is configured ADJANGO_CONTROLLERS_LOGGER_NAME ADJANGO_CONTROLLERS_LOGGING.
  • You can also implement the interface:

    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:

    # settings.py
    from adjango.handlers import HCE # use my example if u need
    ADJANGO_UNCAUGHT_EXCEPTION_HANDLING_FUNCTION = HCE.handle
    

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.

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 ships with extra management commands to speed up project scaffolding.

• astartproject — clones the adjango-template into the given directory and strips its Git history.

  django-admin astartproject myproject
  

• astartup — creates an app skeleton inside apps/ and registers it in INSTALLED_APPS.

  python manage.py astartup blog
  

  After running the command you will have the following structure:

  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.

  python manage.py newentities order apps.commerce Order,Product,Price
  

  Or create a single model:

  python manage.py newentities order apps.commerce Order
  

Other

• AsyncAtomicContextManager🧘

  An asynchronous context manager for working with transactions, which ensures the atomicity of operations.

  from adjango.utils.base import AsyncAtomicContextManager
  
  async def some_function():
      async with AsyncAtomicContextManager():
          ...  
  

• Tasker📋

  The Tasker class provides methods for scheduling tasks in Celery and Celery Beat.

  from adjango.utils.tasks import Tasker
  
  task_id = Tasker.put(
      task=my_celery_task,
      param1='value1',
      param2='value2',
      countdown=60 # The task will be completed in 60 seconds
  )
  

  from adjango.utils.tasks import Tasker
  from datetime import datetime
  
  # One-time task via Celery Beat
  Tasker.beat(
      task=my_celery_task,
      name='one_time_task',
      schedule_time=datetime(2024, 10, 10, 14, 30), # Start the task on October 10, 2024 at 14:30
      param1='value1',
      param2='value2'
  )
  
  # Periodic task via Celery Beat (every hour)
  Tasker.beat(
      task=my_celery_task,
      name='hourly_task',
      interval=3600, # The task runs every hour
      param1='value1',
      param2='value2'
  )
  

• send_emails

  Allows you to send emails using templates and context rendering.

  from adjango.utils.mail import send_emails
  
  send_emails(
      subject='Welcome!',
      emails=('[e-mail]', '[e-mail]'),
      template='emails/welcome.html',
      context={'user': 'John Doe'}
  )
  

  from adjango.tasks import send_emails_task
  from adjango.utils.tasks import Tasker
  
  send_emails_task.delay(
      subject='Hello!',
      emails=('[e-mail]',),
      template='emails/hello.html',
      context={'message': 'Welcome to our service!'}
  )
  # or
  Tasker.put(
      task=send_emails_task,
      subject='Hello!',
      emails=('[e-mail]',),
      template='emails/hello.html',
      context={'message': 'Welcome to our service!'},
      countdown=60 # The task will be completed in 5 seconds
  )