229 lines
8.2 KiB
Python
229 lines
8.2 KiB
Python
from decimal import Decimal
|
|
import json
|
|
|
|
from rest_framework import serializers
|
|
|
|
from apps.notifications.services import notify_workspace_membership_added
|
|
from apps.users.models import User
|
|
from apps.workspaces.services import WORKSPACE_MEMBERS_VIEW, has_workspace_capability
|
|
from core.serializers.base import BaseModelSerializer
|
|
from apps.workspaces.models import PriceUnit, Workspace, WorkspaceMembership, WorkspaceUserRate
|
|
from core.serializers.mini import UserMiniSerializer
|
|
|
|
|
|
class WorkspaceMemberInputSerializer(serializers.Serializer):
|
|
user_id = serializers.UUIDField()
|
|
role = serializers.ChoiceField(choices=WorkspaceMembership.Role.choices, default=WorkspaceMembership.Role.MEMBER)
|
|
|
|
|
|
class WorkspaceSerializer(BaseModelSerializer):
|
|
members = serializers.JSONField(write_only=True, required=False)
|
|
clear_thumbnail = serializers.BooleanField(write_only=True, required=False, default=False)
|
|
my_role = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Workspace
|
|
fields = BaseModelSerializer.Meta.fields + (
|
|
"name",
|
|
"description",
|
|
"thumbnail",
|
|
"clear_thumbnail",
|
|
"owner",
|
|
"my_role",
|
|
"members",
|
|
)
|
|
read_only_fields = BaseModelSerializer.Meta.fields + (
|
|
"owner",
|
|
)
|
|
|
|
def get_my_role(self, obj):
|
|
membership = WorkspaceMembership.objects.filter(
|
|
workspace=obj,
|
|
user=self.context["request"].user,
|
|
).first()
|
|
return getattr(membership, "role", None)
|
|
|
|
def validate_thumbnail(self, value):
|
|
if value is None:
|
|
return value
|
|
max_bytes = 2 * 1024 * 1024
|
|
if getattr(value, "size", 0) > max_bytes:
|
|
raise serializers.ValidationError("Image size must be 2MB or less.")
|
|
content_type = (getattr(value, "content_type", "") or "").lower()
|
|
allowed_types = {"image/jpeg", "image/png", "image/webp"}
|
|
if content_type and content_type not in allowed_types:
|
|
raise serializers.ValidationError("Unsupported image type. Use JPG, PNG, or WebP.")
|
|
return value
|
|
|
|
def to_representation(self, instance):
|
|
data = super().to_representation(instance)
|
|
request = self.context.get("request")
|
|
if instance.thumbnail:
|
|
thumbnail_url = instance.thumbnail.url
|
|
data["thumbnail"] = request.build_absolute_uri(thumbnail_url) if request else thumbnail_url
|
|
else:
|
|
data["thumbnail"] = None
|
|
data.pop("clear_thumbnail", None)
|
|
return data
|
|
|
|
def create(self, validated_data):
|
|
members_data = validated_data.pop("members", [])
|
|
if isinstance(members_data, str):
|
|
try:
|
|
members_data = json.loads(members_data)
|
|
except json.JSONDecodeError as exc:
|
|
raise serializers.ValidationError({"members": "Invalid members format."}) from exc
|
|
if members_data:
|
|
members_serializer = WorkspaceMemberInputSerializer(data=members_data, many=True)
|
|
members_serializer.is_valid(raise_exception=True)
|
|
members_data = members_serializer.validated_data
|
|
validated_data.pop("clear_thumbnail", None)
|
|
|
|
workspace = super().create(validated_data)
|
|
|
|
memberships_to_create = []
|
|
for member in members_data:
|
|
memberships_to_create.append(
|
|
WorkspaceMembership(
|
|
workspace=workspace,
|
|
user_id=member['user_id'],
|
|
role=member['role'],
|
|
is_active=True
|
|
)
|
|
)
|
|
|
|
if memberships_to_create:
|
|
WorkspaceMembership.objects.bulk_create(memberships_to_create)
|
|
request = self.context.get("request")
|
|
actor = getattr(request, "user", None)
|
|
if actor and actor.is_authenticated:
|
|
users_by_id = User.objects.in_bulk(
|
|
[member["user_id"] for member in members_data]
|
|
)
|
|
for member in members_data:
|
|
recipient = users_by_id.get(member["user_id"])
|
|
if recipient:
|
|
notify_workspace_membership_added(
|
|
actor=actor,
|
|
recipient=recipient,
|
|
workspace=workspace,
|
|
role=member["role"],
|
|
)
|
|
|
|
return workspace
|
|
|
|
def update(self, instance, validated_data):
|
|
clear_thumbnail = validated_data.pop("clear_thumbnail", False)
|
|
old_thumbnail_name = instance.thumbnail.name if instance.thumbnail else None
|
|
|
|
if clear_thumbnail and instance.thumbnail:
|
|
instance.thumbnail.delete(save=False)
|
|
instance.thumbnail = None
|
|
|
|
updated_workspace = super().update(instance, validated_data)
|
|
|
|
if old_thumbnail_name and updated_workspace.thumbnail and updated_workspace.thumbnail.name != old_thumbnail_name:
|
|
storage = updated_workspace.thumbnail.storage
|
|
if storage.exists(old_thumbnail_name):
|
|
storage.delete(old_thumbnail_name)
|
|
|
|
return updated_workspace
|
|
|
|
|
|
class WorkspaceMembershipSerializer(BaseModelSerializer):
|
|
user = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = WorkspaceMembership
|
|
fields = BaseModelSerializer.Meta.fields + (
|
|
"workspace",
|
|
"user",
|
|
"role",
|
|
"is_active",
|
|
)
|
|
|
|
def get_user(self, instance):
|
|
request = self.context.get("request")
|
|
viewer = getattr(request, "user", None)
|
|
can_view_sensitive_details = bool(
|
|
viewer
|
|
and viewer.is_authenticated
|
|
and has_workspace_capability(viewer, instance.workspace, WORKSPACE_MEMBERS_VIEW)
|
|
)
|
|
|
|
user_data = UserMiniSerializer(instance.user, context=self.context).data
|
|
if can_view_sensitive_details:
|
|
return user_data
|
|
|
|
return {
|
|
"id": user_data["id"],
|
|
"first_name": user_data.get("first_name"),
|
|
"last_name": user_data.get("last_name"),
|
|
"profile_picture": user_data.get("profile_picture"),
|
|
}
|
|
|
|
|
|
class PriceUnitSerializer(BaseModelSerializer):
|
|
class Meta:
|
|
model = PriceUnit
|
|
fields = BaseModelSerializer.Meta.fields + (
|
|
"code",
|
|
"name",
|
|
"local_name",
|
|
"symbol",
|
|
)
|
|
read_only_fields = fields
|
|
|
|
|
|
class WorkspaceUserRateSerializer(BaseModelSerializer):
|
|
user_details = UserMiniSerializer(source="user", read_only=True)
|
|
price_unit = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = WorkspaceUserRate
|
|
fields = BaseModelSerializer.Meta.fields + (
|
|
"workspace",
|
|
"user",
|
|
"user_details",
|
|
"hourly_rate",
|
|
"currency",
|
|
"price_unit",
|
|
"effective_from",
|
|
)
|
|
read_only_fields = fields
|
|
|
|
def get_price_unit(self, obj):
|
|
unit = PriceUnit.objects.filter(code=obj.currency, is_deleted=False).first()
|
|
if not unit:
|
|
return None
|
|
return PriceUnitSerializer(unit, context=self.context).data
|
|
|
|
|
|
class WorkspaceUserRateCreateSerializer(serializers.Serializer):
|
|
workspace_id = serializers.UUIDField()
|
|
user_id = serializers.UUIDField()
|
|
hourly_rate = serializers.DecimalField(max_digits=10, decimal_places=2, min_value=Decimal("0.01"))
|
|
currency = serializers.CharField(max_length=3, default="USD")
|
|
|
|
def validate_currency(self, value):
|
|
code = value.upper()
|
|
if not PriceUnit.objects.filter(code=code, is_deleted=False).exists():
|
|
raise serializers.ValidationError("Selected price unit is invalid.")
|
|
return code
|
|
|
|
|
|
class WorkspaceUserRateUpdateSerializer(serializers.Serializer):
|
|
hourly_rate = serializers.DecimalField(
|
|
max_digits=10,
|
|
decimal_places=2,
|
|
min_value=Decimal("0.01"),
|
|
required=False,
|
|
)
|
|
currency = serializers.CharField(max_length=3, required=False)
|
|
|
|
def validate_currency(self, value):
|
|
code = value.upper()
|
|
if not PriceUnit.objects.filter(code=code, is_deleted=False).exists():
|
|
raise serializers.ValidationError("Selected price unit is invalid.")
|
|
return code
|