I'm trying to set up a task with Django and Celery. The Celery and Django configuration is okay, nothing to report on that side. However, I have a problem, I think with the writing, of my code in OOP. I can't locate where the problem is. It's an argument problem, 4 arguments are expected by delay(), but my method only expects 3. Here are the files and what I tried to do, as well as the error trace.
My files
helpers.py
I want to create a task for run_test_optimization method
class MachineLearniaXHelpers(MagicModels):
def __init__(self, data_dict, target_name, type_model, test_size = 0.33, random_state = 42, **kwargs) -> None:
super().__init__()
self.test_size = test_size
self.random_state = random_state
self.process = DataProcessor(
data=data_dict,
target_name=target_name
)
self.metrics = MetricsScorer(type_model=type_model)
self.model = self.several_algorythme(type_model, **kwargs)
@staticmethod
def split_data(data_dict):
return train_test_split(*data_dict)
@staticmethod
def train_model(model, X, y):
model.fit(X, y)
def run_test_optimization(self):
dict_result = []
for model in self.model:
feature_model, values = self.process.transform_dict_to_array_structure()
x_train, x_test, y_train, y_test = self.split_data(values)
self.train_model(model, x_train, y_train)
dict_metrics_train = self.metrics.choice_metrics_by_type_model(y_train, model.predict(x_train))
dict_metrics_test = self.metrics.choice_metrics_by_type_model(y_test, model.predict(x_test))
dict_result.append({
"name_model" : model.__class__.__name__,
"features_model" : feature_model,
"train_performances" : dict_metrics_train,
"test_performances" : dict_metrics_test
})
return dict_result
tasks.py
create my task task_run_test_optimization
from celery import shared_task
from .helpers import MachineLearniaXHelpers
@shared_task
def task_run_test_optimization(data_dict, target_name, type_model):
constructor = MachineLearniaXHelpers(
data_dict,
target_name,
type_model
)
dict_result = constructor.run_test_optimization()
return dict_result
views.py
API post method, the task method is here task_run_test_optimization
class MachineLearningXView(APIView):
serializer_class = BuildModelMLSerializer
permission_classes = [IsAuthenticated, UserPermissionMachineLearniaX]
def post(self, request):
if not self.request.session.exists(self.request.session.session_key):
self.request.session.create()
serializer = self.serializer_class(data=request.data)
if serializer.is_valid():
data = request.data.get('data')
target = serializer.data.get('target_name')
type_model = serializer.data.get('test_type')
result = task_run_test_optimization.delay(data, target, type_model) # probleme is here
print(result)
return Response(status=status.HTTP_200_OK)
result send me an id in my console, the post request is ok:
System check identified no issues (0 silenced).
February 08, 2023 - 17:25:28
Django version 4.1.5, using settings 'ialab.settings'
Starting development server at http://127.0.0.1:8000/
Quit the server with CTRL-BREAK.
829e66ee-925f-4aff-8c42-49420e5758ce <-- PRINT result IS HERE
[08/Feb/2023 17:25:47] "POST /api/machinelearniax HTTP/1.1" 200 0
But in my celery worker, i'have this error:
[2023-02-08 17:25:47,813: INFO/MainProcess] Task api.tasks.task_run_test_optimization[829e66ee-925f-4aff-8c42-49420e5758ce] received
[2023-02-08 17:25:47,897: ERROR/MainProcess] Task api.tasks.task_run_test_optimization[829e66ee-925f-4aff-8c42-49420e5758ce] raised unexpected: TypeError('task_run_test_optimization() takes 3 positional arguments but 4 were given')
Traceback (most recent call last):
File "C:\Users\basti\Desktop\IALab\ialab\.ialabenv\lib\site-packages\celery\app\trace.py", line 451, in trace_task
R = retval = fun(*args, **kwargs)
File "C:\Users\basti\Desktop\IALab\ialab\.ialabenv\lib\site-packages\celery\app\trace.py", line 734, in __protected_call__
return self.run(*args, **kwargs)
TypeError: task_run_test_optimization() takes 3 positional arguments but 4 were given
the task is received but not working.
