247 lines
9.0 KiB
Python
247 lines
9.0 KiB
Python
from rest_framework import viewsets, permissions, authentication
|
|
from rest_framework.response import Response
|
|
from jobposting.models import JobOffer, CompanyLogo, Skill
|
|
from core.models import MyUser, AnonymousUserData
|
|
from rest_framework import status
|
|
|
|
from oauth2_provider.contrib.rest_framework import TokenHasReadWriteScope, OAuth2Authentication
|
|
from .auth import OAuth2ClientCredentialAuthentication
|
|
from rest_framework.views import APIView
|
|
|
|
from django.conf import settings
|
|
|
|
from django.db.models import Q
|
|
|
|
from jobposting.serializers import (
|
|
JobOfferSerializer,
|
|
MyUserSerializer,
|
|
CompanyLogoSerializer,
|
|
SkillSerializer,
|
|
SlimJobOffersSerializer,
|
|
MainJobOfferSerializer
|
|
)
|
|
|
|
def remove_none_from_filters(filter: int|str, filter_type: str) -> int|str|None:
|
|
if filter_type == 'int':
|
|
if filter is not None or filter != '':
|
|
try:
|
|
filter = int(filter)
|
|
except ValueError:
|
|
filter = None
|
|
except TypeError:
|
|
filter = None
|
|
else:
|
|
filter = None
|
|
else:
|
|
if filter is not None or filter != '':
|
|
pass
|
|
else:
|
|
filter = None
|
|
|
|
print(filter) # test only
|
|
|
|
return filter
|
|
|
|
|
|
class MyUserViewSet(viewsets.ModelViewSet):
|
|
# authentication_classes = [OAuth2Authentication, authentication.SessionAuthentication]
|
|
# permission_classes = [permissions.IsAuthenticated|TokenHasReadWriteScope]
|
|
queryset = MyUser.objects.all()
|
|
serializer_class = MyUserSerializer
|
|
required_scope = ['main']
|
|
authentication_classes = [] ### test only - to be changed
|
|
permission_classes = [] ###
|
|
# # permission_classes = [permissions.IsAuthenticated]
|
|
|
|
|
|
class JobOfferViewSet(viewsets.ModelViewSet):
|
|
# authentication_classes = [OAuth2Authentication, authentication.SessionAuthentication]
|
|
# permission_classes = [permissions.IsAuthenticated|TokenHasReadWriteScope] # test only
|
|
queryset = JobOffer.objects.all()
|
|
serializer_class = JobOfferSerializer
|
|
required_scope = ['main']
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
# # permission_classes = [permissions.IsAuthenticatedOrReadOnly]
|
|
|
|
def perform_create(self, serializer):
|
|
if self.request.user.is_authenticated:
|
|
serializer.save(created_by=self.request.user)
|
|
else:
|
|
# Pobieranie danych z formularza
|
|
first_name = self.request.data.get('first_name')
|
|
last_name = self.request.data.get('last_name')
|
|
contact_email = self.request.data.get('contact_email')
|
|
|
|
# Tworzenie rekordu AnonymousUserData
|
|
anonymous_user_data = AnonymousUserData.objects.create(
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
contact_email=contact_email
|
|
)
|
|
|
|
serializer.save(anonymous_user_data=anonymous_user_data)
|
|
|
|
|
|
class CompanyLogoViewSet(viewsets.ModelViewSet):
|
|
# authentication_classes = [OAuth2Authentication, authentication.SessionAuthentication]
|
|
# permission_classes = [permissions.IsAuthenticated|TokenHasReadWriteScope]
|
|
queryset = CompanyLogo.objects.all()
|
|
serializer_class = CompanyLogoSerializer
|
|
required_scope = ['main']
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
|
|
class JobOfferListView(APIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
|
|
|
|
def get(self, request, format=None):
|
|
dict_request = dict(request.query_params)
|
|
print(dict_request)
|
|
categories = dict_request.get('categories[]', None)
|
|
max_salary = remove_none_from_filters(
|
|
request.query_params.get('max_salary', None), 'int')
|
|
min_salary = remove_none_from_filters(
|
|
request.query_params.get('min_salary', None), 'int')
|
|
localization = remove_none_from_filters(
|
|
request.query_params.get('localization', None), 'str')
|
|
name = remove_none_from_filters(
|
|
request.query_params.get('name', None), 'str')
|
|
|
|
filters = {
|
|
'name__icontains': name,
|
|
'min_salary__gte': min_salary,
|
|
'max_salary__lte': max_salary,
|
|
'localization': localization,
|
|
'category__in': categories,
|
|
'experience_level': request.query_params.get('experience_level', None),
|
|
'employment_type': request.query_params.get('employment_type', None),
|
|
'work_from_home': request.query_params.get('work_from_home', None),
|
|
}
|
|
|
|
# remove None values
|
|
filters = {k: v for k, v in filters.items() if v is not None}
|
|
print(filters)
|
|
job_offers = JobOffer.objects.filter(**filters)
|
|
serializer = SlimJobOffersSerializer(job_offers, many=True)
|
|
return Response(serializer.data)
|
|
|
|
|
|
class SkillViewset(viewsets.ModelViewSet):
|
|
# authentication_classes = [OAuth2Authentication, authentication.SessionAuthentication]
|
|
# permission_classes = [permissions.IsAuthenticated|TokenHasReadWriteScope]
|
|
queryset = Skill.objects.all()
|
|
serializer_class = SkillSerializer
|
|
required_scope = ['main']
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
|
|
class SkillMainView(APIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
def get_object(self, pk):
|
|
try:
|
|
return Skill.objects.get(pk=pk)
|
|
except Skill.DoesNotExist:
|
|
return Response({'error': 'Not Found'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
def get(self, request, pk=None, format=None):
|
|
if pk:
|
|
skill = self.get_object(pk)
|
|
serializer = SkillSerializer(skill)
|
|
else:
|
|
skill = Skill.objects.all()
|
|
serializer = SkillSerializer(skill, many=True)
|
|
return Response(serializer.data)
|
|
|
|
def post(self, request, format=None):
|
|
serializer = SkillSerializer(data=request.data)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
def put(self, request, pk, format=None):
|
|
skill = self.get_object(pk)
|
|
serializer = SkillSerializer(skill, data=request.data)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return Response(serializer.data)
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
def delete(self, request, pk, format=None):
|
|
skill = self.get_object(pk)
|
|
skill.delete()
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
class JobOfferDetailView(APIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
def get_object(self, pk):
|
|
try:
|
|
return JobOffer.objects.get(pk=pk)
|
|
except JobOffer.DoesNotExist:
|
|
return Response({'error': 'Not Found'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
def get(self, request, pk, format=None):
|
|
job_offer = self.get_object(pk)
|
|
serializer = MainJobOfferSerializer(job_offer,)
|
|
return Response(serializer.data)
|
|
|
|
def put(self, request, pk, format=None):
|
|
job_offer = self.get_object(pk)
|
|
serializer = JobOfferSerializer(job_offer, data=request.data)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return Response(serializer.data)
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
def delete(self, request, pk, format=None):
|
|
job_listing = self.get_object(pk)
|
|
job_listing.delete()
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
def patch(self, request, pk, format=None):
|
|
job_offer = self.get_object(pk)
|
|
serializer = JobOfferSerializer(job_offer, data=request.data, partial=True)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return Response(serializer.data)
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
class JobOfferCreateView(APIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
def post(self, request, format=None):
|
|
print(request.META)
|
|
serializer = JobOfferSerializer(data=request.data)
|
|
if serializer.is_valid():
|
|
if request.user.is_authenticated:
|
|
serializer.save(created_by=request.user)
|
|
else:
|
|
# Pobieranie danych z formularza
|
|
first_name = request.data.get('first_name')
|
|
last_name = request.data.get('last_name')
|
|
contact_email = request.data.get('contact_email')
|
|
|
|
# Tworzenie rekordu AnonymousUserData
|
|
anonymous_user_data = AnonymousUserData.objects.create(
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
contact_email=contact_email
|
|
)
|
|
|
|
serializer.save(anonymous_user_data=anonymous_user_data)
|
|
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
|