forked from erp-dev/erp
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
from typing import Any, Dict, List
|
|
|
|
from celery import shared_task
|
|
|
|
from business import services as business_services
|
|
|
|
|
|
@shared_task(bind=True)
|
|
def create_purchase_order_stock_entries(
|
|
self,
|
|
*,
|
|
purchase_order_id: int,
|
|
warehouse_id: int,
|
|
items: List[Dict[str, Any]],
|
|
created_by_id: int | None = None,
|
|
) -> Dict[str, Any]:
|
|
payload = business_services.create_purchase_order_stock_entries_sync(
|
|
purchase_order_id=purchase_order_id,
|
|
warehouse_id=warehouse_id,
|
|
items=items,
|
|
created_by_id=created_by_id,
|
|
)
|
|
payload['task_id'] = self.request.id
|
|
return payload
|
|
|
|
|
|
@shared_task(bind=True)
|
|
def create_sales_order_stock_entries(
|
|
self,
|
|
*,
|
|
sales_order_id: int,
|
|
warehouse_id: int,
|
|
items: List[Dict[str, Any]],
|
|
created_by_id: int | None = None,
|
|
) -> Dict[str, Any]:
|
|
payload = business_services.create_sales_order_stock_entries_sync(
|
|
sales_order_id=sales_order_id,
|
|
warehouse_id=warehouse_id,
|
|
items=items,
|
|
created_by_id=created_by_id,
|
|
)
|
|
payload['task_id'] = self.request.id
|
|
return payload
|
|
|