# 🚀 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, which offers various useful managers, services, decorators, utilities for asynchronous programming, a task scheduler for Celery, working with transactions and much more.
- [Installation](#installation-%EF%B8%8F)
- [Settings](#settings-%EF%B8%8F)
- [Overview](#overview)
- [Manager & Services](#manager--services-%EF%B8%8F)
- [Utils](#utils-)
- [Decorators](#decorators-)
- [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.polymorphic_models import APolymorphicModel
class User(AbstractUser, ABaseService):
objects = AManager()
# Its equal with...
class User(AbstractUser, AModel): pass
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`, и так далее доступны как отдельные функции
```python
from adjango.utils.funcs import aall, agetorn, afilter, aset, aadd, arelated
```
### 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.
```python
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:
```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
```
### 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():
...
```
* `Tasker`📋
The Tasker class provides methods for scheduling tasks in `Celery` and `Celery Beat`.
```python
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
)
```
```python
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.
```python
from adjango.utils.mail import send_emails
send_emails(
subject='Welcome!',
emails=('user1@example.com', 'user2@example.com'),
template='emails/welcome.html',
context={'user': 'John Doe'}
)
```
```python
from adjango.tasks import send_emails_task
from adjango.utils.tasks import Tasker
send_emails_task.delay(
subject='Hello!',
emails=('user@example.com',),
template='emails/hello.html',
context={'message': 'Welcome to our service!'}
)
# or
Tasker.put(
task=send_emails_task,
subject='Hello!',
emails=('user@example.com',),
template='emails/hello.html',
context={'message': 'Welcome to our service!'},
countdown=60 # The task will be completed in 5 seconds
)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/Artasov/adjango",
"name": "adjango",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"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/d6/b0/5c9c1468df05f959c21fe1387707f942f821f31217d9fe49f9004a6faf78/adjango-0.3.1.tar.gz",
"platform": null,
"description": "# \ud83d\ude80 ADjango \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 convenient library for simplifying work with Django, which offers various useful managers, services, decorators, utilities for asynchronous programming, a task scheduler for Celery, working with transactions and much more.\r\n- [Installation](#installation-%EF%B8%8F)\r\n- [Settings](#settings-%EF%B8%8F)\r\n- [Overview](#overview)\r\n - [Manager & Services](#manager--services-%EF%B8%8F)\r\n - [Utils](#utils-)\r\n - [Decorators](#decorators-)\r\n - [Other](#other)\r\n\r\n## Installation \ud83d\udee0\ufe0f\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 ```python\r\n INSTALLED_APPS = [\r\n #...\r\n 'adjango',\r\n ]\r\n ```\r\n* ### In `settings.py` set the params\r\n ```python\r\n # settings.py\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 ```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## Overview\r\nMost functions, if available in asynchronous form, are also available in synchronous form.\r\n\r\n### Manager & Services \ud83d\udece\ufe0f\r\nA simple example and everything is immediately clear...\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.polymorphic_models import APolymorphicModel\r\n\r\nclass User(AbstractUser, ABaseService):\r\n objects = AManager()\r\n# Its equal with...\r\nclass User(AbstractUser, AModel): pass\r\n \r\n\r\nclass Product(APolymorphicModel):\r\n # APolymorphicManager() of course here already exists\r\n name = CharField(max_length=100)\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.related('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### Utils \ud83d\udd27\r\n `aall`, `afilter`, `arelated`, \u0438 \u0442\u0430\u043a \u0434\u0430\u043b\u0435\u0435 \u0434\u043e\u0441\u0442\u0443\u043f\u043d\u044b \u043a\u0430\u043a \u043e\u0442\u0434\u0435\u043b\u044c\u043d\u044b\u0435 \u0444\u0443\u043d\u043a\u0446\u0438\u0438\r\n ```python\r\n from adjango.utils.funcs import aall, agetorn, afilter, aset, aadd, arelated\r\n ```\r\n### Decorators \ud83c\udf80\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* `atomic`\r\n\r\n An asynchronous decorator that wraps \r\n function into a transactional context. If an exception occurs, all changes are rolled back.\r\n\r\n* `acontroller/controller`\r\n\r\n An asynchronous decorator that wraps \r\n function into a transactional context. If an exception occurs, all changes are rolled back.\r\n ```python\r\n from adjango.adecorators import acontroller\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 * These decorators automatically catch uncaught exceptions and log if the logger is configured \r\n `ADJANGO_CONTROLLERS_LOGGER_NAME` `ADJANGO_CONTROLLERS_LOGGING`. \r\n * You can also implement the interface:\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 and use `handle` to get an uncaught exception:\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### 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 ```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\r\n* `Tasker`\ud83d\udccb\r\n\r\n The Tasker class provides methods for scheduling tasks in `Celery` and `Celery Beat`.\r\n ```python\r\n from adjango.utils.tasks import Tasker\r\n \r\n task_id = Tasker.put(\r\n task=my_celery_task,\r\n param1='value1',\r\n param2='value2',\r\n countdown=60 # The task will be completed in 60 seconds\r\n )\r\n ```\r\n ```python\r\n from adjango.utils.tasks import Tasker\r\n from datetime import datetime\r\n \r\n # One-time task via Celery Beat\r\n Tasker.beat(\r\n task=my_celery_task,\r\n name='one_time_task',\r\n schedule_time=datetime(2024, 10, 10, 14, 30), # Start the task on October 10, 2024 at 14:30\r\n param1='value1',\r\n param2='value2'\r\n )\r\n \r\n # Periodic task via Celery Beat (every hour)\r\n Tasker.beat(\r\n task=my_celery_task,\r\n name='hourly_task',\r\n interval=3600, # The task runs every hour\r\n param1='value1',\r\n param2='value2'\r\n )\r\n ```\r\n\r\n* `send_emails`\r\n\r\n Allows you to send emails using templates and context rendering.\r\n ```python\r\n from adjango.utils.mail import send_emails\r\n \r\n send_emails(\r\n subject='Welcome!',\r\n emails=('user1@example.com', 'user2@example.com'),\r\n template='emails/welcome.html',\r\n context={'user': 'John Doe'}\r\n )\r\n ```\r\n ```python\r\n from adjango.tasks import send_emails_task\r\n from adjango.utils.tasks import Tasker\r\n \r\n send_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 # or\r\n Tasker.put(\r\n task=send_emails_task,\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 countdown=60 # The task will be completed in 5 seconds\r\n )\r\n ```\r\n",
"bugtrack_url": null,
"license": null,
"summary": "A library with many features for interacting with Django",
"version": "0.3.1",
"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": "",
"digests": {
"blake2b_256": "3f4bbb8e66d44a9b5cb50a381c30e26161cf2af97cf8ea7572557b9eb8008e6e",
"md5": "4585da4e209e73429d4d2f8f3c35748b",
"sha256": "7db0920a10eb7537dfd2e7acd9a90d0b813505d699c61d7c400fc5ffe4e395b4"
},
"downloads": -1,
"filename": "adjango-0.3.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "4585da4e209e73429d4d2f8f3c35748b",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 38987,
"upload_time": "2024-12-13T11:30:30",
"upload_time_iso_8601": "2024-12-13T11:30:30.092500Z",
"url": "https://files.pythonhosted.org/packages/3f/4b/bb8e66d44a9b5cb50a381c30e26161cf2af97cf8ea7572557b9eb8008e6e/adjango-0.3.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "d6b05c9c1468df05f959c21fe1387707f942f821f31217d9fe49f9004a6faf78",
"md5": "1b2d3c184c30ae9074e00720613ffc33",
"sha256": "f03c19d4a94ad8dfe04381b0e8ec34112110505fb263e42649c28472f0d684b2"
},
"downloads": -1,
"filename": "adjango-0.3.1.tar.gz",
"has_sig": false,
"md5_digest": "1b2d3c184c30ae9074e00720613ffc33",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 30018,
"upload_time": "2024-12-13T11:30:32",
"upload_time_iso_8601": "2024-12-13T11:30:32.808774Z",
"url": "https://files.pythonhosted.org/packages/d6/b0/5c9c1468df05f959c21fe1387707f942f821f31217d9fe49f9004a6faf78/adjango-0.3.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-13 11:30:32",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Artasov",
"github_project": "adjango",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [],
"lcname": "adjango"
}
JSON
