Coverage for app/venv/lib/python3.14/site-packages/weblate/api/views.py: 69%

1448 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 07:15 +0000

1# Copyright © Michal Čihař <michal@weblate.org> 

2# SPDX-FileCopyrightText: 2025 Javier Pérez <jdbp@protonmail.com> 

3# 

4# SPDX-License-Identifier: GPL-3.0-or-later 

5 

6from __future__ import annotations 

7 

8import os.path 

9from datetime import datetime 

10from typing import TYPE_CHECKING, cast 

11from urllib.parse import unquote 

12 

13from celery.result import AsyncResult 

14from django.conf import settings 

15from django.contrib.messages import get_messages 

16from django.core.cache import cache 

17from django.core.exceptions import PermissionDenied 

18from django.db import transaction 

19from django.db.models import Q 

20from django.forms.utils import from_current_timezone 

21from django.http import FileResponse, Http404 

22from django.shortcuts import get_object_or_404 

23from django.utils.datastructures import MultiValueDictKeyError 

24from django.utils.html import format_html 

25from django.utils.translation import gettext, gettext_lazy 

26from django_filters import rest_framework as filters 

27from drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_view 

28from drf_standardized_errors.handler import ExceptionHandler 

29from rest_framework import parsers, viewsets 

30from rest_framework.decorators import action 

31from rest_framework.exceptions import APIException, ValidationError 

32from rest_framework.mixins import CreateModelMixin, DestroyModelMixin, UpdateModelMixin 

33from rest_framework.permissions import IsAuthenticated 

34from rest_framework.response import Response 

35from rest_framework.reverse import reverse 

36from rest_framework.settings import api_settings 

37from rest_framework.status import ( 

38 HTTP_200_OK, 

39 HTTP_201_CREATED, 

40 HTTP_204_NO_CONTENT, 

41 HTTP_400_BAD_REQUEST, 

42 HTTP_423_LOCKED, 

43 HTTP_500_INTERNAL_SERVER_ERROR, 

44) 

45from rest_framework.utils import formatting 

46from rest_framework.views import APIView 

47from rest_framework.viewsets import ViewSet 

48 

49from weblate.accounts.models import Subscription 

50from weblate.accounts.utils import remove_user 

51from weblate.addons.models import Addon 

52from weblate.api.pagination import LargePagination 

53from weblate.api.serializers import ( 

54 AddonSerializer, 

55 BasicUserSerializer, 

56 BilingualSourceUnitSerializer, 

57 BilingualUnitSerializer, 

58 CategorySerializer, 

59 ChangeSerializer, 

60 CommentSerializer, 

61 ComponentListSerializer, 

62 ComponentSerializer, 

63 FullUserSerializer, 

64 GroupSerializer, 

65 LabelSerializer, 

66 LanguageSerializer, 

67 LockRequestSerializer, 

68 LockSerializer, 

69 MemorySerializer, 

70 MetricsSerializer, 

71 MonolingualUnitSerializer, 

72 NotificationSerializer, 

73 ProjectLockSerializer, 

74 ProjectMachinerySettingsSerializer, 

75 ProjectSerializer, 

76 RepoRequestSerializer, 

77 RoleSerializer, 

78 ScreenshotCreateSerializer, 

79 ScreenshotFileSerializer, 

80 ScreenshotSerializer, 

81 SingleServiceConfigSerializer, 

82 StatisticsSerializer, 

83 TranslationSerializer, 

84 UnitSerializer, 

85 UnitWriteSerializer, 

86 UploadRequestSerializer, 

87 UserStatisticsSerializer, 

88 edit_service_settings_response_serializer, 

89 get_reverse_kwargs, 

90) 

91from weblate.auth.models import Group, Role, User 

92from weblate.auth.results import PermissionResult 

93from weblate.formats.models import EXPORTERS 

94from weblate.lang.models import Language 

95from weblate.machinery.models import validate_service_configuration 

96from weblate.memory.models import Memory 

97from weblate.screenshots.models import Screenshot 

98from weblate.trans.actions import ActionEvents 

99from weblate.trans.autotranslate import AutoTranslate 

100from weblate.trans.exceptions import FileParseError 

101from weblate.trans.forms import AutoForm 

102from weblate.trans.models import ( 

103 Category, 

104 Change, 

105 Component, 

106 ComponentList, 

107 Label, 

108 Project, 

109 Unit, 

110) 

111from weblate.trans.models.translation import Translation, TranslationQuerySet 

112from weblate.trans.tasks import ( 

113 category_removal, 

114 component_removal, 

115 project_removal, 

116) 

117from weblate.trans.views.files import download_multi 

118from weblate.trans.views.reports import generate_credits 

119from weblate.utils.celery import get_task_progress 

120from weblate.utils.docs import get_doc_url 

121from weblate.utils.errors import report_error 

122from weblate.utils.lock import WeblateLockTimeoutError 

123from weblate.utils.search import parse_query 

124from weblate.utils.state import ( 

125 STATE_APPROVED, 

126 STATE_EMPTY, 

127 STATE_FUZZY, 

128 STATE_TRANSLATED, 

129) 

130from weblate.utils.stats import GlobalStats, prefetch_stats 

131from weblate.utils.views import download_translation_file, zip_download 

132 

133from .renderers import FlatJsonRenderer, OpenMetricsRenderer 

134 

135if TYPE_CHECKING: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true

136 from django.db.models import Model 

137 from django.http import HttpResponse 

138 from rest_framework.request import Request 

139 

140 from weblate.api.serializers import ( 

141 NewUnitSerializer, 

142 ) 

143 from weblate.auth.models import AuthenticatedHttpRequest 

144 

145REPO_OPERATIONS: dict[str, tuple[str, str, tuple, dict, bool]] = { 

146 "push": ("vcs.push", "do_push", (), {}, True), 

147 "pull": ("vcs.update", "do_update", (), {}, True), 

148 "pull-rebase": ("vcs.update", "do_update", (), {"method": "rebase"}, True), 

149 "pull-merge": ("vcs.update", "do_update", (), {"method": "merge"}, True), 

150 "pull-merge-noff": ("vcs.update", "do_update", (), {"method": "merge_noff"}, True), 

151 "reset": ("vcs.reset", "do_reset", (), {}, True), 

152 "reset-keep": ("vcs.reset", "do_reset", (), {"keep_changes": True}, True), 

153 "cleanup": ("vcs.reset", "do_cleanup", (), {}, True), 

154 "commit": ("vcs.commit", "commit_pending", ("api",), {}, False), 

155 "file-sync": ("vcs.reset", "do_file_sync", (), {}, True), 

156 "file-scan": ("vcs.reset", "do_file_scan", (), {}, True), 

157} 

158 

159DOC_TEXT = """ 

160<p>See <a href="{0}">the Weblate's Web API documentation</a> for detailed 

161description of the API.</p> 

162""" 

163 

164 

165class LockedError(APIException): 

166 status_code = HTTP_423_LOCKED 

167 default_detail = gettext_lazy("Could not obtain the lock to perform the operation.") 

168 default_code = "unknown-locked" 

169 

170 

171class NotSourceUnit(APIException): 

172 status_code = HTTP_400_BAD_REQUEST 

173 default_detail = gettext_lazy("Specified unit id is not a translation source unit.") 

174 default_code = "not-a-source-unit" 

175 

176 

177class WeblateExceptionHandler(ExceptionHandler): 

178 def convert_known_exceptions(self, exc: Exception) -> Exception: 

179 if isinstance(exc, WeblateLockTimeoutError): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true

180 if exc.lock.scope == "repo": 

181 return LockedError( 

182 code="repository-locked", 

183 detail=gettext( 

184 "Could not obtain the repository lock for %s to perform the operation." 

185 ) 

186 % exc.lock.origin, 

187 ) 

188 if exc.lock.scope == "component-update": 

189 return LockedError( 

190 code="component-locked", 

191 detail=gettext( 

192 "Could not obtain the update lock for component %s to perform the operation." 

193 ) 

194 % exc.lock.origin, 

195 ) 

196 if exc.lock.origin: 

197 return LockedError( 

198 detail=gettext( 

199 "Could not obtain the %(scope)s lock for %(origin)s to perform the operation." 

200 ) 

201 % {"scope": exc.lock.scope, "origin": exc.lock.origin}, 

202 ) 

203 return LockedError() 

204 return super().convert_known_exceptions(exc) 

205 

206 

207def get_view_description(view, html=False): 

208 """ 

209 Given a view class, return a textual description to represent the view. 

210 

211 This name is used in the browsable API, and in OPTIONS responses. This function is 

212 the default for the `VIEW_DESCRIPTION_FUNCTION` setting. 

213 """ 

214 description = view.__doc__ or "" 

215 description = formatting.dedent(description) 

216 

217 if hasattr(getattr(view, "serializer_class", "None"), "Meta"): 

218 model_name = view.serializer_class.Meta.model.__name__.lower() 

219 doc_name = "categories" if model_name == "category" else f"{model_name}s" 

220 doc_url = get_doc_url("api", doc_name, user=view.request.user) 

221 else: 

222 doc_url = get_doc_url("api", user=view.request.user) 

223 

224 if html: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true

225 return formatting.markup_description(description) + format_html( 

226 DOC_TEXT, doc_url 

227 ) 

228 return description 

229 

230 

231class DownloadViewSet(viewsets.ReadOnlyModelViewSet): 

232 raw_urls: tuple[str, ...] = () 

233 raw_formats: tuple[str, ...] = tuple(EXPORTERS) 

234 

235 def perform_content_negotiation(self, request: Request, force=False): 

236 """Perform custom content negotiation.""" 

237 if ( 

238 request.resolver_match is not None 

239 and request.resolver_match.url_name in self.raw_urls 

240 ): 

241 fmt = self.format_kwarg 

242 if fmt is None or fmt in self.raw_formats: 242 ↛ 245line 242 didn't jump to line 245 because the condition on line 242 was always true

243 renderers = self.get_renderers() 

244 return (renderers[0], renderers[0].media_type) 

245 msg = "Not supported format" 

246 raise Http404(msg) 

247 return super().perform_content_negotiation(request, force) 

248 

249 def download_file( 

250 self, filename: str, content_type: str, component: Component | None = None 

251 ) -> HttpResponse: 

252 """Download file.""" 

253 if os.path.isdir(filename): 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true

254 return zip_download( 

255 filename, 

256 [filename], 

257 name=component.slug if component else "weblate", 

258 ) 

259 try: 

260 open_file = open(filename, "rb") # noqa: SIM115 

261 except FileNotFoundError as error: 

262 msg = "File not found" 

263 raise Http404(msg) from error 

264 return FileResponse( 

265 open_file, 

266 content_type=content_type, 

267 as_attachment=True, 

268 filename=os.path.basename(filename), 

269 ) 

270 

271 

272class WeblateViewSet(DownloadViewSet): 

273 """Allow to skip content negotiation for certain requests.""" 

274 

275 def repository_operation( 

276 self, request: Request, obj, project: Project, operation: str 

277 ): 

278 permission, method, args, kwargs, takes_request = REPO_OPERATIONS[operation] 

279 

280 if not request.user.has_perm(permission, project): 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 raise PermissionDenied 

282 

283 obj.acting_user = request.user 

284 

285 if takes_request: 

286 return getattr(obj, method)(*args, request, **kwargs) 

287 return getattr(obj, method)(*args, request.user, **kwargs) 

288 

289 @extend_schema( 

290 description="Return information about VCS repository status.", methods=["get"] 

291 ) 

292 @extend_schema( 

293 description="Perform given operation on the VCS repository.", methods=["post"] 

294 ) 

295 @action( 

296 detail=True, methods=["get", "post"], serializer_class=RepoRequestSerializer 

297 ) 

298 def repository(self, request: Request, **kwargs): 

299 obj = self.get_object() 

300 

301 if isinstance(obj, Translation): 

302 project = obj.component.project 

303 elif isinstance(obj, Component): 

304 project = obj.project 

305 else: 

306 project = obj 

307 

308 if request.method == "POST": 

309 serializer = RepoRequestSerializer(data=request.data) 

310 serializer.is_valid(raise_exception=True) 

311 

312 data = { 

313 "result": self.repository_operation( 

314 request, obj, project, serializer.validated_data["operation"] 

315 ) 

316 } 

317 

318 storage = get_messages(request) 

319 if storage: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true

320 data["detail"] = "\n".join(m.message for m in storage) 

321 

322 return Response(data) 

323 

324 if not request.user.has_perm("meta:vcs.status", project): 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true

325 raise PermissionDenied 

326 

327 data = { 

328 "needs_commit": obj.needs_commit(), 

329 "needs_merge": obj.repo_needs_merge(), 

330 "needs_push": obj.repo_needs_push(), 

331 } 

332 

333 if isinstance(obj, Project): 

334 data["url"] = reverse( 

335 "api:project-repository", kwargs={"slug": obj.slug}, request=request 

336 ) 

337 else: 

338 if isinstance(obj, Translation): 

339 component = obj.component 

340 data["url"] = reverse( 

341 "api:translation-repository", 

342 kwargs=get_reverse_kwargs( 

343 obj, 

344 ( 

345 "component__project__slug", 

346 "component__slug", 

347 "language__code", 

348 ), 

349 ), 

350 request=request, 

351 ) 

352 else: 

353 component = obj 

354 data["url"] = reverse( 

355 "api:component-repository", 

356 kwargs=get_reverse_kwargs(obj, ("project__slug", "slug")), 

357 request=request, 

358 ) 

359 

360 data["remote_commit"] = component.get_last_remote_commit() 

361 data["weblate_commit"] = component.get_last_commit() 

362 data["status"] = component.repository.status() 

363 changes = component.change_set.filter( 

364 action__in=Change.ACTIONS_REPOSITORY 

365 ).order_by("-id") 

366 

367 if changes.exists() and changes[0].is_merge_failure(): 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true

368 data["merge_failure"] = changes[0].target 

369 else: 

370 data["merge_failure"] = None 

371 

372 return Response(data) 

373 

374 

375class MultipleFieldViewSet(WeblateViewSet): 

376 """ 

377 Multiple field filtering mixin. 

378 

379 Apply this mixin to any view or viewset to get multiple field filtering based on a 

380 `lookup_fields` attribute, instead of the default single field filtering. 

381 """ 

382 

383 def get_object(self): 

384 # Get the base queryset 

385 queryset = self.get_queryset() 

386 # Apply any filter backends 

387 queryset = self.filter_queryset(queryset) 

388 # Generate lookup 

389 lookup = {} 

390 category_path = "" 

391 for field in reversed(self.lookup_fields): 

392 if field not in {"component__slug", "slug"}: 

393 lookup[field] = self.kwargs[field] 

394 else: 

395 category_prefix = field[:-4] 

396 was_category = False 

397 was_slug = False 

398 # Fetch component part for possible category 

399 for category in reversed(unquote(self.kwargs[field]).split("/")): 

400 if not was_slug: 400 ↛ 406line 400 didn't jump to line 406 because the condition on line 400 was always true

401 # Component filter 

402 lookup[field] = category 

403 was_slug = True 

404 else: 

405 # Strip "slug" from category field 

406 category_path = f"category__{category_path}" 

407 lookup[f"{category_prefix}{category_path}slug"] = category 

408 was_category = True 

409 if not was_category: 409 ↛ 391line 409 didn't jump to line 391 because the condition on line 409 was always true

410 # No category 

411 lookup[f"{category_prefix}category"] = None 

412 

413 # Lookup the object 

414 return get_object_or_404(queryset, **lookup) 

415 

416 

417class UserFilter(filters.FilterSet): 

418 username = filters.CharFilter(field_name="username", lookup_expr="startswith") 

419 

420 class Meta: 

421 model = User 

422 fields = ("username", "id") 

423 

424 

425@extend_schema_view( 

426 retrieve=extend_schema(description="Return information about users."), 

427 partial_update=extend_schema(description="Change the user parameters."), 

428) 

429class UserViewSet(viewsets.ModelViewSet): 

430 """Users API.""" 

431 

432 queryset = User.objects.none() 

433 lookup_field = "username" 

434 filter_backends = (filters.DjangoFilterBackend,) 

435 filterset_class = UserFilter 

436 

437 def get_serializer_class(self): 

438 if self.request.user.has_perm("user.view") or self.request.user.has_perm( 

439 "user.edit" 

440 ): 

441 return FullUserSerializer 

442 return BasicUserSerializer 

443 

444 def get_queryset(self): 

445 user = self.request.user 

446 if not user.is_authenticated: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true

447 return User.objects.none() 

448 queryset = User.objects.order_by("id") 

449 if not user.has_perm("user.edit") and not user.has_perm("user.view"): 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true

450 return queryset 

451 return queryset.prefetch_related("groups", "profile", "profile__languages") 

452 

453 def list(self, request, *args, **kwargs): 

454 """ 

455 List of users if you have permissions to see manage users. 

456 

457 Without a permission you get to see only your own details. 

458 """ 

459 # Copy of rest_framework.mixins.ListModelMixin.list with additional 

460 # filtering based on user permissions. We limit listing of user to 

461 # non-admins, but access to individual users is retained (with limited 

462 # serializer). 

463 user = self.request.user 

464 if not user.is_authenticated: 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true

465 queryset = User.objects.none() 

466 elif ( 466 ↛ 471line 466 didn't jump to line 471 because the condition on line 466 was never true

467 not (user.has_perm("user.edit") or user.has_perm("user.view")) 

468 and self.lookup_field not in request.GET 

469 ): 

470 # Order to avoid warning from the paginator 

471 queryset = User.objects.filter(pk=user.pk).order_by("id") 

472 else: 

473 queryset = self.get_queryset() 

474 

475 queryset = self.filter_queryset(queryset) 

476 

477 page = self.paginate_queryset(queryset) 

478 if page is not None: 478 ↛ 482line 478 didn't jump to line 482 because the condition on line 478 was always true

479 serializer = self.get_serializer(page, many=True) 

480 return self.get_paginated_response(serializer.data) 

481 

482 serializer = self.get_serializer(queryset, many=True) 

483 return Response(serializer.data) 

484 

485 def perm_check(self, request: Request) -> None: 

486 if not request.user.has_perm("user.edit"): 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true

487 self.permission_denied(request, "Can not manage Users") 

488 

489 def update(self, request: Request, *args, **kwargs): 

490 """Change the user parameters.""" 

491 self.perm_check(request) 

492 return super().update(request, *args, **kwargs) 

493 

494 def create(self, request: Request, *args, **kwargs): 

495 """Create a new user.""" 

496 self.perm_check(request) 

497 return super().create(request, *args, **kwargs) 

498 

499 def destroy(self, request: Request, *args, **kwargs): 

500 """Delete all user information and mark the user inactive.""" 

501 self.perm_check(request) 

502 instance = self.get_object() 

503 remove_user(instance, cast("AuthenticatedHttpRequest", request)) 

504 return Response(status=HTTP_204_NO_CONTENT) 

505 

506 @extend_schema(description="Associate groups with a user.", methods=["post"]) 

507 @extend_schema(description="Remove a user from a group.", methods=["delete"]) 

508 @action(detail=True, methods=["post", "delete"]) 

509 def groups(self, request: Request, **kwargs): 

510 obj = self.get_object() 

511 self.perm_check(request) 

512 

513 if "group_id" not in request.data: 513 ↛ 517line 513 didn't jump to line 517 because the condition on line 513 was always true

514 msg = "Missing group_id parameter" 

515 raise ValidationError({"group_id": msg}) 

516 

517 try: 

518 group = Group.objects.get(pk=int(request.data["group_id"])) 

519 except (Group.DoesNotExist, ValueError) as error: 

520 raise ValidationError({"group_id": str(error)}) from error 

521 

522 if request.method == "POST": 

523 obj.add_team(request, group) 

524 if request.method == "DELETE": 

525 obj.remove_team(request, group) 

526 serializer = self.get_serializer_class()(obj, context={"request": request}) 

527 

528 return Response(serializer.data, status=HTTP_200_OK) 

529 

530 @extend_schema(description="List subscriptions of a user.", methods=["get"]) 

531 @extend_schema(description="Associate subscriptions with a user.", methods=["post"]) 

532 @extend_schema( 

533 request=NotificationSerializer, 

534 responses=NotificationSerializer(many=True), 

535 ) 

536 @action( 

537 detail=True, 

538 methods=["get", "post"], 

539 serializer_class=NotificationSerializer(many=True), 

540 ) 

541 def notifications(self, request: Request, username: str): 

542 obj = self.get_object() 

543 if request.method == "POST": 

544 self.perm_check(request) 

545 with transaction.atomic(): 

546 serializer = NotificationSerializer( 

547 data=request.data, context={"request": request} 

548 ) 

549 serializer.is_valid(raise_exception=True) 

550 serializer.save(user=obj) 

551 return Response(serializer.data, status=HTTP_201_CREATED) 

552 queryset = obj.subscription_set.order_by("id") 

553 page = self.paginate_queryset(queryset) 

554 serializer = NotificationSerializer( 

555 page, many=True, context={"request": request} 

556 ) 

557 return self.get_paginated_response(serializer.data) 

558 

559 @extend_schema( 

560 description="Get a subscription associated with a user.", methods=["get"] 

561 ) 

562 @extend_schema( 

563 description="Edit a subscription associated with a user.", 

564 methods=["put", "patch"], 

565 ) 

566 @extend_schema( 

567 description="Delete a subscription associated with a user.", methods=["delete"] 

568 ) 

569 @extend_schema( 

570 parameters=[ 

571 OpenApiParameter("subscription_id", int, OpenApiParameter.PATH), 

572 ], 

573 responses=NotificationSerializer, 

574 request=NotificationSerializer, 

575 ) 

576 @action( 

577 detail=True, 

578 methods=["get", "put", "patch", "delete"], 

579 url_path="notifications/(?P<subscription_id>[0-9]+)", 

580 serializer_class=NotificationSerializer, 

581 ) 

582 def notifications_details( 

583 self, request: Request, username: str, subscription_id: int 

584 ): 

585 obj = self.get_object() 

586 

587 try: 

588 subscription = obj.subscription_set.get(id=subscription_id) 

589 except Subscription.DoesNotExist as error: 

590 raise Http404(str(error)) from error 

591 

592 if request.method == "DELETE": 

593 self.perm_check(request) 

594 subscription.delete() 

595 return Response(status=HTTP_204_NO_CONTENT) 

596 

597 if request.method == "GET": 

598 serializer = NotificationSerializer( 

599 subscription, context={"request": request} 

600 ) 

601 else: 

602 self.perm_check(request) 

603 serializer = NotificationSerializer( 

604 subscription, 

605 data=request.data, 

606 context={"request": request}, 

607 partial=request.method == "PATCH", 

608 ) 

609 serializer.is_valid(raise_exception=True) 

610 serializer.save() 

611 

612 return Response(serializer.data, status=HTTP_200_OK) 

613 

614 @extend_schema( 

615 description="List statistics of a user.", 

616 methods=["get"], 

617 tags=["users", "statistics"], 

618 ) 

619 @action( 

620 detail=True, 

621 methods=["get"], 

622 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

623 ) 

624 def statistics(self, request: Request, **kwargs): 

625 obj = self.get_object() 

626 

627 serializer = UserStatisticsSerializer(obj, context={"request": request}) 

628 

629 return Response(serializer.data) 

630 

631 @extend_schema( 

632 description="List translation contributions of a user.", 

633 methods=["get"], 

634 tags=["users", "contributions"], 

635 ) 

636 @action( 

637 detail=True, 

638 methods=["get"], 

639 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

640 ) 

641 def contributions(self, request: Request, **kwargs): 

642 user_translation_ids = set( 

643 Change.objects.content() 

644 .filter(user=self.get_object()) 

645 .values_list("translation", flat=True) 

646 ) 

647 user_translations = ( 

648 Translation.objects.filter_access(request.user) 

649 .prefetch() 

650 .filter( 

651 id__in=user_translation_ids, 

652 ) 

653 .order() 

654 ) 

655 

656 page = self.paginate_queryset(user_translations) 

657 

658 serializer = TranslationSerializer( 

659 page, 

660 many=True, 

661 context={"request": request}, 

662 ) 

663 

664 return self.get_paginated_response(serializer.data) 

665 

666 

667@extend_schema_view( 

668 create=extend_schema(description="Create a new group."), 

669 retrieve=extend_schema(description="Return information about a group."), 

670 partial_update=extend_schema(description="Change the group parameters."), 

671) 

672class GroupViewSet(viewsets.ModelViewSet): 

673 """Groups API.""" 

674 

675 queryset = Group.objects.none() 

676 serializer_class = GroupSerializer 

677 lookup_field = "id" 

678 

679 def get_queryset(self): 

680 user = self.request.user 

681 if user.has_perm("group.edit") or user.has_perm("group.view"): 681 ↛ 684line 681 didn't jump to line 684 because the condition on line 681 was always true

682 queryset = Group.objects.all() 

683 else: 

684 queryset = Group.objects.filter(Q(user=user) | Q(admins=user)).distinct() 

685 return queryset.order_by("id") 

686 

687 def perm_check( 

688 self, 

689 request: Request, 

690 group: Group | None = None, 

691 project: Project | None = None, 

692 ) -> None: 

693 # if a project is provided and the user has the required permission to edit teams in the project, allow access 

694 if project is not None and request.user.has_perm("meta:team.edit", project): 

695 return 

696 

697 if (group is None and not self.request.user.has_perm("group.edit")) or ( 697 ↛ 700line 697 didn't jump to line 700 because the condition on line 697 was never true

698 group is not None and not request.user.has_perm("meta:team.edit", group) 

699 ): 

700 self.permission_denied(request, "Can not manage groups") 

701 

702 def update(self, request: Request, *args, **kwargs): 

703 """Change the group parameters.""" 

704 self.perm_check(request) 

705 return super().update(request, *args, **kwargs) 

706 

707 def perform_create(self, serializer) -> None: 

708 self.perm_check( 

709 self.request, project=serializer.validated_data.get("defining_project") 

710 ) 

711 super().perform_create(serializer) 

712 

713 def destroy(self, request: Request, *args, **kwargs): 

714 """Delete the group.""" 

715 self.perm_check(request) 

716 return super().destroy(request, *args, **kwargs) 

717 

718 @extend_schema(description="Associate roles with a group.", methods=["post"]) 

719 @action(detail=True, methods=["post"]) 

720 def roles(self, request: Request, **kwargs): 

721 obj = self.get_object() 

722 self.perm_check(request) 

723 

724 if "role_id" not in request.data: 724 ↛ 728line 724 didn't jump to line 728 because the condition on line 724 was always true

725 msg = "Missing role_id parameter" 

726 raise ValidationError({"role_id": msg}) 

727 

728 try: 

729 role = Role.objects.get(pk=int(request.data["role_id"])) 

730 except (Role.DoesNotExist, ValueError) as error: 

731 raise ValidationError({"role_id": str(error)}) from error 

732 

733 obj.roles.add(role) 

734 serializer = self.serializer_class(obj, context={"request": request}) 

735 

736 return Response(serializer.data, status=HTTP_200_OK) 

737 

738 @extend_schema( 

739 description="Delete a role from a group.", 

740 methods=["delete"], 

741 parameters=[OpenApiParameter("role_id", int, OpenApiParameter.PATH)], 

742 ) 

743 @action(detail=True, methods=["delete"], url_path="roles/(?P<role_id>[0-9]+)") 

744 def delete_roles(self, request: Request, id, role_id): # noqa: A002 

745 obj = self.get_object() 

746 self.perm_check(request) 

747 

748 try: 

749 role = obj.roles.get(pk=role_id) 

750 except Role.DoesNotExist as error: 

751 raise Http404(str(error)) from error 

752 

753 obj.roles.remove(role) 

754 return Response(status=HTTP_204_NO_CONTENT) 

755 

756 @extend_schema(description="Associate languages with a group.", methods=["post"]) 

757 @action( 

758 detail=True, 

759 methods=["post"], 

760 ) 

761 def languages(self, request: Request, **kwargs): 

762 obj = self.get_object() 

763 self.perm_check(request) 

764 

765 if "language_code" not in request.data: 765 ↛ 769line 765 didn't jump to line 769 because the condition on line 765 was always true

766 msg = "Missing language_code parameter" 

767 raise ValidationError({"language_code": msg}) 

768 

769 try: 

770 language = Language.objects.get(code=request.data["language_code"]) 

771 except (Language.DoesNotExist, ValueError) as error: 

772 raise ValidationError({"language_code": str(error)}) from error 

773 

774 obj.languages.add(language) 

775 serializer = self.serializer_class(obj, context={"request": request}) 

776 

777 return Response(serializer.data, status=HTTP_200_OK) 

778 

779 @extend_schema( 

780 description="Delete a language from a group.", 

781 methods=["delete"], 

782 parameters=[OpenApiParameter("language_code", str, OpenApiParameter.PATH)], 

783 ) 

784 @action( 

785 detail=True, methods=["delete"], url_path="languages/(?P<language_code>[^/.]+)" 

786 ) 

787 def delete_languages(self, request: Request, id, language_code): # noqa: A002 

788 obj = self.get_object() 

789 self.perm_check(request) 

790 

791 try: 

792 language = obj.languages.get(code=language_code) 

793 except Language.DoesNotExist as error: 

794 raise Http404(str(error)) from error 

795 obj.languages.remove(language) 

796 return Response(status=HTTP_204_NO_CONTENT) 

797 

798 @extend_schema(description="Associate projects with a group.", methods=["post"]) 

799 @action( 

800 detail=True, 

801 methods=["post"], 

802 ) 

803 def projects(self, request: Request, **kwargs): 

804 obj = self.get_object() 

805 self.perm_check(request) 

806 

807 if "project_id" not in request.data: 807 ↛ 811line 807 didn't jump to line 811 because the condition on line 807 was always true

808 msg = "Missing project_id parameter" 

809 raise ValidationError({"project_id": msg}) 

810 

811 try: 

812 project = Project.objects.get( 

813 pk=int(request.data["project_id"]), 

814 ) 

815 except (Project.DoesNotExist, ValueError) as error: 

816 raise ValidationError({"project_id": str(error)}) from error 

817 obj.projects.add(project) 

818 serializer = self.serializer_class(obj, context={"request": request}) 

819 

820 return Response(serializer.data, status=HTTP_200_OK) 

821 

822 @extend_schema(description="Delete a project from a group.", methods=["delete"]) 

823 @action(detail=True, methods=["delete"], url_path="projects/(?P<project_id>[0-9]+)") 

824 def delete_projects(self, request: Request, id, project_id): # noqa: A002 

825 obj = self.get_object() 

826 self.perm_check(request) 

827 

828 try: 

829 project = obj.projects.get(pk=project_id) 

830 except Project.DoesNotExist as error: 

831 raise Http404(str(error)) from error 

832 obj.projects.remove(project) 

833 return Response(status=HTTP_204_NO_CONTENT) 

834 

835 @extend_schema( 

836 description="Associate componentlists with a group.", methods=["post"] 

837 ) 

838 @action(detail=True, methods=["post"]) 

839 def componentlists(self, request: Request, **kwargs): 

840 obj = self.get_object() 

841 self.perm_check(request) 

842 

843 if "component_list_id" not in request.data: 843 ↛ 847line 843 didn't jump to line 847 because the condition on line 843 was always true

844 msg = "Missing component_list_id parameter" 

845 raise ValidationError({"component_list_id": msg}) 

846 

847 try: 

848 component_list = ComponentList.objects.get( 

849 pk=int(request.data["component_list_id"]), 

850 ) 

851 except (ComponentList.DoesNotExist, ValueError) as error: 

852 raise ValidationError({"component_list_id": str(error)}) from error 

853 obj.componentlists.add(component_list) 

854 serializer = self.serializer_class(obj, context={"request": request}) 

855 

856 return Response(serializer.data, status=HTTP_200_OK) 

857 

858 @extend_schema( 

859 description="Delete a componentlist from a group.", methods=["delete"] 

860 ) 

861 @action( 

862 detail=True, 

863 methods=["delete"], 

864 url_path="componentlists/(?P<component_list_id>[0-9]+)", 

865 ) 

866 def delete_componentlists( 

867 self, 

868 request: Request, 

869 id, # noqa: A002 

870 component_list_id, 

871 ): 

872 obj = self.get_object() 

873 self.perm_check(request) 

874 try: 

875 component_list = obj.componentlists.get(pk=component_list_id) 

876 except ComponentList.DoesNotExist as error: 

877 raise Http404(str(error)) from error 

878 obj.componentlists.remove(component_list) 

879 return Response(status=HTTP_204_NO_CONTENT) 

880 

881 @extend_schema(description="Associate components with a group.", methods=["post"]) 

882 @action( 

883 detail=True, 

884 methods=["post"], 

885 ) 

886 def components(self, request: Request, **kwargs): 

887 obj = self.get_object() 

888 self.perm_check(request) 

889 if "component_id" not in request.data: 889 ↛ 893line 889 didn't jump to line 893 because the condition on line 889 was always true

890 msg = "Missing component_id parameter" 

891 raise ValidationError({"component_id": msg}) 

892 

893 try: 

894 component = Component.objects.filter_access(request.user).get( 

895 pk=int(request.data["component_id"]) 

896 ) 

897 except (Component.DoesNotExist, ValueError) as error: 

898 raise ValidationError({"component_id": str(error)}) from error 

899 obj.components.add(component) 

900 serializer = self.serializer_class(obj, context={"request": request}) 

901 

902 return Response(serializer.data, status=HTTP_200_OK) 

903 

904 @extend_schema(description="Delete a component from a group.", methods=["delete"]) 

905 @action( 

906 detail=True, methods=["delete"], url_path="components/(?P<component_id>[0-9]+)" 

907 ) 

908 def delete_components(self, request: Request, id, component_id): # noqa: A002 

909 obj = self.get_object() 

910 self.perm_check(request) 

911 

912 try: 

913 component = obj.components.get(pk=component_id) 

914 except Component.DoesNotExist as error: 

915 raise Http404(str(error)) from error 

916 obj.components.remove(component) 

917 return Response(status=HTTP_204_NO_CONTENT) 

918 

919 @extend_schema(description="Make user a group admin.", methods=["post"]) 

920 @action(detail=True, methods=["post"], url_path="admins") 

921 def grant_admin(self, request: Request, id): # noqa: A002 

922 group = self.get_object() 

923 if not request.user.has_perm("meta:team.users", group): 923 ↛ 924line 923 didn't jump to line 924 because the condition on line 923 was never true

924 self.perm_check(request, group) 

925 user_id = request.data.get("user_id") 

926 if not user_id: 926 ↛ 930line 926 didn't jump to line 930 because the condition on line 926 was always true

927 msg = "User ID is required" 

928 raise ValidationError({"user_id": msg}) 

929 

930 try: 

931 user = User.objects.get(pk=user_id) 

932 except User.DoesNotExist as error: 

933 msg = "User not found" 

934 raise ValidationError({"user_id": msg}) from error 

935 group.admins.add(user) 

936 user.add_team(cast("AuthenticatedHttpRequest", request), group) 

937 return Response({"Administration rights granted."}, status=HTTP_200_OK) 

938 

939 @extend_schema(description="Delete a user from group admins.", methods=["delete"]) 

940 @action(detail=True, methods=["delete"], url_path="admins/(?P<user_pk>[0-9]+)") 

941 def revoke_admin(self, request: Request, id, user_pk): # noqa: A002 

942 group = self.get_object() 

943 if not request.user.has_perm("meta:team.users", group): 

944 self.perm_check(request, group) 

945 try: 

946 user = group.admins.get(pk=user_pk) # Using user_pk from the URL path 

947 except User.DoesNotExist as error: 

948 msg = "User not found" 

949 raise ValidationError(msg) from error 

950 

951 group.admins.remove(user) 

952 serializer = GroupSerializer(group, context={"request": request}) 

953 return Response(serializer.data, status=HTTP_200_OK) 

954 

955 

956@extend_schema_view( 

957 list=extend_schema( 

958 description="Return a list of all roles associated with the user." 

959 ), 

960 retrieve=extend_schema(description="Return information about a role."), 

961 partial_update=extend_schema(description="Change the role parameters."), 

962) 

963class RoleViewSet(viewsets.ModelViewSet): 

964 """Roles API.""" 

965 

966 queryset = Role.objects.none() 

967 serializer_class = RoleSerializer 

968 lookup_field = "id" 

969 

970 def get_queryset(self): 

971 user = self.request.user 

972 if user.has_perm("role.edit") or user.has_perm("role.view"): 972 ↛ 974line 972 didn't jump to line 974 because the condition on line 972 was always true

973 return Role.objects.order_by("id").all() 

974 return ( 

975 Role.objects.filter(group__in=user.groups.all()).order_by("id").distinct() 

976 ) 

977 

978 def perm_check(self, request: Request) -> None: 

979 if not request.user.has_perm("role.edit"): 979 ↛ 980line 979 didn't jump to line 980 because the condition on line 979 was never true

980 self.permission_denied(request, "Can not manage roles") 

981 

982 def update(self, request: Request, *args, **kwargs): 

983 """Change the role parameters.""" 

984 self.perm_check(request) 

985 return super().update(request, *args, **kwargs) 

986 

987 def create(self, request: Request, *args, **kwargs): 

988 """Create a new role.""" 

989 self.perm_check(request) 

990 return super().create(request, *args, **kwargs) 

991 

992 def destroy(self, request: Request, *args, **kwargs): 

993 """Delete a role.""" 

994 self.perm_check(request) 

995 return super().destroy(request, *args, **kwargs) 

996 

997 

998class CreditsMixin: 

999 @action(detail=True, methods=["get"]) 

1000 def credits(self, request, **kwargs): 

1001 if request.user.is_anonymous: 1001 ↛ 1002line 1001 didn't jump to line 1002 because the condition on line 1001 was never true

1002 self.permission_denied(request, "Must be authenticated to get credits") 

1003 

1004 obj = self.get_object() 

1005 

1006 try: 

1007 start_date = from_current_timezone( 

1008 datetime.fromisoformat(request.query_params["start"]) 

1009 ) 

1010 except (ValueError, MultiValueDictKeyError) as err: 

1011 msg = "Invalid format for `start`" 

1012 raise ValidationError({"start": msg}) from err 

1013 

1014 try: 

1015 end_date = from_current_timezone( 

1016 datetime.fromisoformat(request.query_params["end"]) 

1017 ) 

1018 except (ValueError, MultiValueDictKeyError) as err: 

1019 msg = "Invalid format for `end`" 

1020 raise ValidationError({"end": msg}) from err 

1021 

1022 language = None 

1023 

1024 if "lang" in request.query_params: 

1025 language = request.query_params["lang"] 

1026 

1027 data = generate_credits( 

1028 None if request.user.has_perm("reports.view", obj) else request.user, 

1029 start_date, 

1030 end_date, 

1031 language, 

1032 obj, 

1033 "count", 

1034 "descending", 

1035 ) 

1036 return Response(data=data) 

1037 

1038 

1039@extend_schema_view( 

1040 list=extend_schema(description="Return a list of all projects."), 

1041 retrieve=extend_schema(description="Return information about a project."), 

1042 partial_update=extend_schema(description="Edit a project by a PATCH request."), 

1043 credits=extend_schema(description="Return contributor credits for a project."), 

1044) 

1045class ProjectViewSet( 

1046 WeblateViewSet, UpdateModelMixin, CreateModelMixin, DestroyModelMixin, CreditsMixin 

1047): 

1048 """Translation projects API.""" 

1049 

1050 raw_urls: tuple[str, ...] = "project-file" 

1051 raw_formats = ("zip", *(f"zip:{exporter}" for exporter in EXPORTERS)) 

1052 

1053 queryset = Project.objects.none() 

1054 serializer_class = ProjectSerializer 

1055 lookup_field = "slug" 

1056 request: Request # type: ignore[assignment] 

1057 

1058 def get_queryset(self): 

1059 return self.request.user.allowed_projects.prefetch_related( 

1060 "addon_set" 

1061 ).order_by("id") 

1062 

1063 @extend_schema( 

1064 description="Return a list of translation components in the given project.", 

1065 methods=["get"], 

1066 ) 

1067 @extend_schema( 

1068 description="Create translation components in the given project.", 

1069 methods=["post"], 

1070 ) 

1071 @action( 

1072 detail=True, 

1073 methods=["get", "post"], 

1074 parser_classes=( 

1075 parsers.JSONParser, 

1076 parsers.MultiPartParser, 

1077 parsers.FormParser, 

1078 parsers.FileUploadParser, 

1079 ), 

1080 serializer_class=ComponentSerializer, 

1081 ) 

1082 def components(self, request: Request, **kwargs): 

1083 obj = self.get_object() 

1084 if request.method == "POST": 

1085 if not request.user.has_perm("project.edit", obj): 1085 ↛ 1086line 1085 didn't jump to line 1086 because the condition on line 1085 was never true

1086 self.permission_denied(request, "Can not create components") 

1087 with transaction.atomic(): 

1088 serializer = ComponentSerializer( 

1089 data=request.data, context={"request": request, "project": obj} 

1090 ) 

1091 serializer.is_valid(raise_exception=True) 

1092 serializer.save() 

1093 serializer.instance.post_create(self.request.user, origin="api") 

1094 return Response( 

1095 serializer.data, 

1096 status=HTTP_201_CREATED, 

1097 headers={ 

1098 "Location": str(serializer.data[api_settings.URL_FIELD_NAME]) 

1099 }, 

1100 ) 

1101 

1102 queryset = obj.component_set.filter_access(self.request.user).order_by("id") 

1103 page = self.paginate_queryset(queryset) 

1104 

1105 serializer = ComponentSerializer( 

1106 page, many=True, context={"request": request}, remove_fields=("project",) 

1107 ) 

1108 

1109 return self.get_paginated_response(serializer.data) 

1110 

1111 @extend_schema(description="Return categories for a project.", methods=["get"]) 

1112 @action(detail=True, methods=["get"]) 

1113 def categories(self, request: Request, **kwargs): 

1114 obj = self.get_object() 

1115 

1116 queryset = obj.category_set.order_by("id") 

1117 page = self.paginate_queryset(queryset) 

1118 

1119 serializer = CategorySerializer(page, many=True, context={"request": request}) 

1120 

1121 return self.get_paginated_response(serializer.data) 

1122 

1123 @extend_schema( 

1124 description="Return statistics for a project.", 

1125 methods=["get"], 

1126 tags=["projects", "statistics"], 

1127 ) 

1128 @action( 

1129 detail=True, 

1130 methods=["get"], 

1131 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

1132 ) 

1133 def statistics(self, request: Request, **kwargs): 

1134 obj = self.get_object() 

1135 

1136 serializer = StatisticsSerializer(obj, context={"request": request}) 

1137 

1138 return Response(serializer.data) 

1139 

1140 @extend_schema( 

1141 description="Return paginated statistics for all languages within a project.", 

1142 methods=["get"], 

1143 tags=["projects", "statistics"], 

1144 ) 

1145 @action( 

1146 detail=True, 

1147 methods=["get"], 

1148 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

1149 ) 

1150 def languages(self, request: Request, **kwargs): 

1151 obj = self.get_object() 

1152 

1153 serializer = StatisticsSerializer( 

1154 obj.stats.get_language_stats(), many=True, context={"request": request} 

1155 ) 

1156 

1157 return Response(serializer.data) 

1158 

1159 @extend_schema(description="Return a list of project changes.", methods=["get"]) 

1160 @action(detail=True, methods=["get"]) 

1161 def changes(self, request: Request, **kwargs): 

1162 obj = self.get_object() 

1163 

1164 queryset = obj.change_set.prefetch().order() 

1165 queryset = ChangesFilterBackend().filter_queryset(request, queryset, self) 

1166 page = self.paginate_queryset(queryset) 

1167 

1168 serializer = ChangeSerializer(page, many=True, context={"request": request}) 

1169 

1170 return self.get_paginated_response(serializer.data) 

1171 

1172 @extend_schema(description="Return labels for a project.", methods=["get"]) 

1173 @extend_schema(description="Create a label for a project.", methods=["post"]) 

1174 @action(detail=True, methods=["get", "post"]) 

1175 def labels(self, request: Request, **kwargs): 

1176 obj = self.get_object() 

1177 

1178 if request.method == "POST": 

1179 if not request.user.has_perm("project.edit", obj): 1179 ↛ 1180line 1179 didn't jump to line 1180 because the condition on line 1179 was never true

1180 self.permission_denied(request, "Can not create labels") 

1181 with transaction.atomic(): 

1182 serializer = LabelSerializer( 

1183 data=request.data, context={"request": request, "project": obj} 

1184 ) 

1185 serializer.is_valid(raise_exception=True) 

1186 serializer.save(project=obj) 

1187 return Response( 

1188 serializer.data, 

1189 status=HTTP_201_CREATED, 

1190 ) 

1191 

1192 # GET request - return all labels for the project 

1193 queryset = obj.label_set.all().order_by("id") 

1194 page = self.paginate_queryset(queryset) 

1195 serializer = LabelSerializer(page, many=True, context={"request": request}) 

1196 return self.get_paginated_response(serializer.data) 

1197 

1198 @extend_schema( 

1199 description="Delete a label from a project.", 

1200 methods=["delete"], 

1201 parameters=[OpenApiParameter("label_id", int, OpenApiParameter.PATH)], 

1202 ) 

1203 @action(detail=True, methods=["delete"], url_path="labels/(?P<label_id>[0-9]+)") 

1204 @transaction.atomic 

1205 def delete_labels(self, request: Request, slug, label_id): 

1206 obj = self.get_object() 

1207 

1208 if not request.user.has_perm("project.edit", obj): 1208 ↛ 1209line 1208 didn't jump to line 1209 because the condition on line 1208 was never true

1209 self.permission_denied(request, "Can not delete labels") 

1210 

1211 try: 

1212 label = obj.label_set.get(id=label_id) 

1213 except Label.DoesNotExist as error: 

1214 msg = f"Label with ID {label_id} was not found in project {obj}" 

1215 raise Http404(msg) from error 

1216 

1217 label.delete() 

1218 

1219 return Response(status=HTTP_204_NO_CONTENT) 

1220 

1221 @action(detail=True, methods=["post"]) 

1222 def addons(self, request: Request, **kwargs): 

1223 obj = self.get_object() 

1224 obj.acting_user = request.user 

1225 

1226 if not request.user.has_perm("project.edit", obj): 1226 ↛ 1227line 1226 didn't jump to line 1227 because the condition on line 1226 was never true

1227 self.permission_denied(request, "Can not create addon") 

1228 

1229 serializer = AddonSerializer( 

1230 data=request.data, context={"request": request, "project": obj} 

1231 ) 

1232 serializer.is_valid(raise_exception=True) 

1233 serializer.save(project=obj) 

1234 return Response(serializer.data, status=HTTP_201_CREATED) 

1235 

1236 def create(self, request: Request, *args, **kwargs): 

1237 """Create a new project.""" 

1238 billing = None 

1239 if not request.user.has_perm("project.add"): 1239 ↛ 1240line 1239 didn't jump to line 1240 because the condition on line 1239 was never true

1240 if "weblate.billing" in settings.INSTALLED_APPS: 

1241 from weblate.billing.models import Billing 

1242 

1243 try: 

1244 billing = Billing.objects.for_user_within_limits(self.request.user)[ 

1245 0 

1246 ] 

1247 except IndexError: 

1248 self.permission_denied( 

1249 request, "No valid billing found or limit exceeded." 

1250 ) 

1251 else: 

1252 self.permission_denied(request, "Can not create projects") 

1253 self.request = request 

1254 self.billing = billing 

1255 return super().create(request, *args, **kwargs) 

1256 

1257 def perform_create(self, serializer) -> None: 

1258 with transaction.atomic(): 

1259 super().perform_create(serializer) 

1260 billing = getattr(self, "billing", None) 

1261 serializer.instance.post_create(self.request.user, billing) 

1262 

1263 def update(self, request: Request, *args, **kwargs): 

1264 """Edit a project by a PUT request.""" 

1265 instance = self.get_object() 

1266 if not request.user.has_perm("project.edit", instance): 1266 ↛ 1267line 1266 didn't jump to line 1267 because the condition on line 1266 was never true

1267 self.permission_denied(request, "Can not edit project") 

1268 instance.acting_user = request.user 

1269 return super().update(request, *args, **kwargs) 

1270 

1271 def destroy(self, request: Request, *args, **kwargs): 

1272 """Delete a project.""" 

1273 instance = self.get_object() 

1274 if not request.user.has_perm("project.edit", instance): 

1275 self.permission_denied(request, "Can not delete project") 

1276 instance.acting_user = request.user 

1277 project_removal.delay(instance.pk, request.user.pk) 

1278 return Response(status=HTTP_204_NO_CONTENT) 

1279 

1280 @action(detail=True, methods=["get"]) 

1281 def file(self, request: Request, **kwargs): 

1282 instance = self.get_object() 

1283 

1284 if not request.user.has_perm("translation.download", instance): 1284 ↛ 1285line 1284 didn't jump to line 1285 because the condition on line 1284 was never true

1285 raise PermissionDenied 

1286 

1287 components = instance.component_set.filter_access(request.user) 

1288 requested_format = request.query_params.get("format", "zip") 

1289 requested_language = request.query_params.get("language_code", None) 

1290 

1291 if requested_language: 1291 ↛ 1292line 1291 didn't jump to line 1292 because the condition on line 1291 was never true

1292 translations = Translation.objects.filter( 

1293 language__code=requested_language, component__in=components 

1294 ) 

1295 else: 

1296 translations = Translation.objects.filter(component__in=components) 

1297 

1298 return download_multi( 

1299 cast("AuthenticatedHttpRequest", request), 

1300 translations.prefetch(), 

1301 [instance], 

1302 requested_format, 

1303 name=instance.slug, 

1304 ) 

1305 

1306 @extend_schema( 

1307 responses=ProjectMachinerySettingsSerializer, 

1308 methods=["GET"], 

1309 description="List machinery settings for a project.", 

1310 ) 

1311 @extend_schema( 

1312 request=SingleServiceConfigSerializer, 

1313 responses=edit_service_settings_response_serializer("post", 201, 400), 

1314 methods=["POST"], 

1315 description="Install a new machinery service.", 

1316 ) 

1317 @extend_schema( 

1318 request=SingleServiceConfigSerializer, 

1319 responses=edit_service_settings_response_serializer("patch", 200, 400), 

1320 methods=["PATCH"], 

1321 description="Partially update a single service. Leave configuration blank to remove the service.", 

1322 ) 

1323 @extend_schema( 

1324 request=ProjectMachinerySettingsSerializer, 

1325 responses=edit_service_settings_response_serializer("put", 200, 400), 

1326 methods=["PUT"], 

1327 description="Replace configuration for all services.", 

1328 ) 

1329 @action(detail=True, methods=["get", "post", "patch", "put"]) 

1330 def machinery_settings(self, request: Request, **kwargs): 

1331 """List or create/update machinery configuration for a project.""" 

1332 project = self.get_object() 

1333 

1334 if not request.user.has_perm("project.edit", project): 1334 ↛ 1335line 1334 didn't jump to line 1335 because the condition on line 1334 was never true

1335 self.permission_denied( 

1336 request, "Can not retrieve/edit machinery configuration" 

1337 ) 

1338 

1339 if request.method in {"POST", "PATCH"}: 

1340 try: 

1341 service_name = request.data["service"] 

1342 except KeyError as error: 

1343 raise ValidationError({"service": "Missing service name"}) from error 

1344 

1345 service, configuration, errors = validate_service_configuration( 

1346 service_name, request.data.get("configuration", "{}") 

1347 ) 

1348 

1349 if service is None or errors: 1349 ↛ 1352line 1349 didn't jump to line 1352 because the condition on line 1349 was always true

1350 raise ValidationError({"configuration": errors}) 

1351 

1352 if request.method == "PATCH": 

1353 if configuration: 

1354 # update a configuration 

1355 project.machinery_settings[service_name] = configuration 

1356 project.save(update_fields=["machinery_settings"]) 

1357 return Response( 

1358 {"message": f"Service updated: {service.name}"}, 

1359 status=HTTP_200_OK, 

1360 ) 

1361 # remove a configuration 

1362 project.machinery_settings.pop(service_name, None) 

1363 project.save(update_fields=["machinery_settings"]) 

1364 return Response( 

1365 {"message": f"Service removed: {service.name}"}, 

1366 status=HTTP_200_OK, 

1367 ) 

1368 

1369 if request.method == "POST": 

1370 if service_name in project.machinery_settings: 

1371 raise ValidationError({"service": ["Service already exists"]}) 

1372 

1373 project.machinery_settings[service_name] = configuration 

1374 project.save(update_fields=["machinery_settings"]) 

1375 return Response( 

1376 {"message": f"Service installed: {service.name}"}, 

1377 status=HTTP_201_CREATED, 

1378 ) 

1379 

1380 elif request.method == "PUT": 

1381 # replace all service configuration 

1382 valid_configurations: dict[str, dict] = {} 

1383 for service_name, configuration in request.data.items(): 

1384 service, configuration, errors = validate_service_configuration( 

1385 service_name, configuration 

1386 ) 

1387 

1388 if service is None or errors: 1388 ↛ 1391line 1388 didn't jump to line 1391 because the condition on line 1388 was always true

1389 raise ValidationError({"configuration": errors}) 

1390 

1391 valid_configurations[service_name] = configuration 

1392 

1393 project.machinery_settings = valid_configurations 

1394 project.save(update_fields=["machinery_settings"]) 

1395 return Response( 

1396 { 

1397 "message": f"Services installed: {', '.join(valid_configurations.keys())}" 

1398 }, 

1399 status=HTTP_201_CREATED, 

1400 ) 

1401 

1402 # GET method 

1403 return Response( 

1404 data=ProjectMachinerySettingsSerializer(project).data, 

1405 status=HTTP_200_OK, 

1406 ) 

1407 

1408 @extend_schema(description="Return project lock status.", methods=["get"]) 

1409 @extend_schema(description="Sets project lock status.", methods=["post"]) 

1410 @action( 

1411 detail=True, methods=["get", "post"], serializer_class=LockRequestSerializer 

1412 ) 

1413 def lock(self, request: Request, **kwargs): 

1414 obj = self.get_object() 

1415 

1416 if request.method == "POST": 

1417 if not request.user.has_perm("project.edit", obj): 1417 ↛ 1418line 1417 didn't jump to line 1418 because the condition on line 1417 was never true

1418 raise PermissionDenied 

1419 

1420 serializer = LockRequestSerializer(data=request.data) 

1421 serializer.is_valid(raise_exception=True) 

1422 

1423 obj.do_lock(request.user, serializer.validated_data["lock"]) 

1424 

1425 return Response(data=ProjectLockSerializer(obj).data) 

1426 

1427 

1428@extend_schema_view( 

1429 list=extend_schema(description="Return a list of translation components."), 

1430 retrieve=extend_schema( 

1431 description="Return information about translation component." 

1432 ), 

1433 partial_update=extend_schema(description="Edit a component by a PATCH request."), 

1434) 

1435class ComponentViewSet( 

1436 MultipleFieldViewSet, UpdateModelMixin, DestroyModelMixin, CreditsMixin 

1437): 

1438 """Translation components API.""" 

1439 

1440 raw_urls: tuple[str, ...] = ("component-file",) 

1441 raw_formats = ("zip", *(f"zip:{exporter}" for exporter in EXPORTERS)) 

1442 

1443 queryset = Component.objects.none() 

1444 serializer_class = ComponentSerializer 

1445 lookup_fields = ("project__slug", "slug") 

1446 

1447 def get_queryset(self): 

1448 return ( 

1449 Component.objects.prefetch(defer=False) 

1450 .filter_access(self.request.user) 

1451 .prefetch_related("source_language", "addon_set") 

1452 .order_by("id") 

1453 ) 

1454 

1455 @extend_schema(description="Return component lock status.", methods=["get"]) 

1456 @extend_schema(description="Sets component lock status.", methods=["post"]) 

1457 @action( 

1458 detail=True, methods=["get", "post"], serializer_class=LockRequestSerializer 

1459 ) 

1460 def lock(self, request: Request, **kwargs): 

1461 obj = self.get_object() 

1462 

1463 if request.method == "POST": 

1464 if not request.user.has_perm("component.lock", obj): 1464 ↛ 1465line 1464 didn't jump to line 1465 because the condition on line 1464 was never true

1465 raise PermissionDenied 

1466 

1467 serializer = LockRequestSerializer(data=request.data) 

1468 serializer.is_valid(raise_exception=True) 

1469 

1470 obj.do_lock(request.user, serializer.validated_data["lock"]) 

1471 

1472 return Response(data=LockSerializer(obj).data) 

1473 

1474 @extend_schema( 

1475 description="Download base file for monolingual translations.", methods=["get"] 

1476 ) 

1477 @action(detail=True, methods=["get"]) 

1478 def monolingual_base(self, request: Request, **kwargs): 

1479 obj = self.get_object() 

1480 

1481 if not obj.has_template(): 1481 ↛ 1485line 1481 didn't jump to line 1485 because the condition on line 1481 was always true

1482 msg = "No template found!" 

1483 raise Http404(msg) 

1484 

1485 return self.download_file( 

1486 obj.get_template_filename(), obj.template_store.mimetype(), component=obj 

1487 ) 

1488 

1489 @extend_schema( 

1490 description="Download template file for new translations.", methods=["get"] 

1491 ) 

1492 @action(detail=True, methods=["get"]) 

1493 def new_template(self, request: Request, **kwargs): 

1494 obj = self.get_object() 

1495 

1496 if not obj.new_base: 

1497 msg = "No file found!" 

1498 raise Http404(msg) 

1499 

1500 return self.download_file(obj.get_new_base_filename(), "application/binary") 

1501 

1502 @extend_schema( 

1503 description="Return a list of translation objects in the given component.", 

1504 methods=["get"], 

1505 ) 

1506 @extend_schema( 

1507 description="Create a new translation in the given component.", methods=["post"] 

1508 ) 

1509 @action(detail=True, methods=["get", "post"]) 

1510 def translations(self, request: Request, **kwargs): 

1511 obj = self.get_object() 

1512 

1513 if request.method == "POST": 

1514 if not request.user.has_perm("translation.add", obj): 

1515 self.permission_denied(request, "Can not create translation") 

1516 

1517 if "language_code" not in request.data: 1517 ↛ 1521line 1517 didn't jump to line 1521 because the condition on line 1517 was always true

1518 msg = "Missing 'language_code' parameter" 

1519 raise ValidationError({"language_code": msg}) 

1520 

1521 language_code = request.data["language_code"] 

1522 

1523 if not obj.can_add_new_language(request.user): 

1524 self.permission_denied(request, message=obj.new_lang_error_message) 

1525 

1526 base_languages = obj.get_all_available_languages() 

1527 if not request.user.has_perm("translation.add_more", obj): 

1528 base_languages = base_languages.filter_for_add(obj.project) 

1529 

1530 try: 

1531 language = base_languages.get(code=language_code) 

1532 except Language.DoesNotExist as error: 

1533 message = f"Could not add {language_code!r}!" 

1534 raise ValidationError({"language_code": message}) from error 

1535 

1536 translation = obj.add_new_language(language, request) 

1537 if translation is None: 

1538 storage = get_messages(request) 

1539 if storage: 

1540 message = "\n".join(m.message for m in storage) 

1541 else: 

1542 message = f"Could not add {language_code!r}!" 

1543 raise ValidationError({"language_code": message}) 

1544 

1545 serializer = TranslationSerializer( 

1546 translation, context={"request": request}, remove_fields=("component",) 

1547 ) 

1548 

1549 return Response(data={"data": serializer.data}, status=HTTP_201_CREATED) 

1550 

1551 queryset = obj.translation_set.prefetch().prefetch_plurals().order_by("id") 

1552 page = self.paginate_queryset(queryset) 

1553 

1554 # Prefetch workflow settings 

1555 if page: 1555 ↛ 1558line 1555 didn't jump to line 1558 because the condition on line 1555 was always true

1556 page[0].component.project.project_languages.preload_workflow_settings() 

1557 

1558 serializer = TranslationSerializer( 

1559 prefetch_stats(page), 

1560 many=True, 

1561 context={"request": request}, 

1562 remove_fields=("component",), 

1563 ) 

1564 

1565 return self.get_paginated_response(serializer.data) 

1566 

1567 @action(detail=True, methods=["post"]) 

1568 def addons(self, request: Request, **kwargs): 

1569 obj = self.get_object() 

1570 obj.acting_user = request.user 

1571 

1572 if not request.user.has_perm("component.edit", obj): 1572 ↛ 1573line 1572 didn't jump to line 1573 because the condition on line 1572 was never true

1573 self.permission_denied(request, "Can not create addon") 

1574 

1575 serializer = AddonSerializer( 

1576 data=request.data, context={"request": request, "component": obj} 

1577 ) 

1578 serializer.is_valid(raise_exception=True) 

1579 serializer.save(component=obj) 

1580 return Response(serializer.data, status=HTTP_201_CREATED) 

1581 

1582 @extend_schema( 

1583 description="Return paginated statistics for all translations within component.", 

1584 methods=["get"], 

1585 tags=["components", "statistics"], 

1586 ) 

1587 @action( 

1588 detail=True, 

1589 methods=["get"], 

1590 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

1591 ) 

1592 def statistics(self, request: Request, **kwargs): 

1593 obj = self.get_object() 

1594 

1595 queryset = obj.translation_set.all().prefetch_meta().order_by("id") 

1596 

1597 paginator = LargePagination() 

1598 page = paginator.paginate_queryset(queryset, request, view=self) 

1599 

1600 serializer = StatisticsSerializer( 

1601 prefetch_stats(page), many=True, context={"request": request} 

1602 ) 

1603 

1604 return paginator.get_paginated_response(serializer.data) 

1605 

1606 @extend_schema(description="Return a list of component changes.", methods=["get"]) 

1607 @action(detail=True, methods=["get"]) 

1608 def changes(self, request: Request, **kwargs): 

1609 obj = self.get_object() 

1610 

1611 queryset = obj.change_set.prefetch().order() 

1612 queryset = ChangesFilterBackend().filter_queryset(request, queryset, self) 

1613 page = self.paginate_queryset(queryset) 

1614 

1615 serializer = ChangeSerializer(page, many=True, context={"request": request}) 

1616 

1617 return self.get_paginated_response(serializer.data) 

1618 

1619 @extend_schema( 

1620 description="Return a list of component screenshots.", methods=["get"] 

1621 ) 

1622 @action(detail=True, methods=["get"]) 

1623 def screenshots(self, request: Request, **kwargs): 

1624 obj = self.get_object() 

1625 

1626 queryset = Screenshot.objects.filter(translation__component=obj).order_by("id") 

1627 page = self.paginate_queryset(queryset) 

1628 

1629 serializer = ScreenshotSerializer(page, many=True, context={"request": request}) 

1630 

1631 return self.get_paginated_response(serializer.data) 

1632 

1633 def update(self, request: Request, *args, **kwargs): 

1634 """Edit a component by a PUT request.""" 

1635 instance = self.get_object() 

1636 if not request.user.has_perm("component.edit", instance): 1636 ↛ 1637line 1636 didn't jump to line 1637 because the condition on line 1636 was never true

1637 self.permission_denied(request, "Can not edit component") 

1638 instance.acting_user = request.user 

1639 return super().update(request, *args, **kwargs) 

1640 

1641 def destroy(self, request: Request, *args, **kwargs): 

1642 """Delete a component.""" 

1643 instance = self.get_object() 

1644 if not request.user.has_perm("component.edit", instance): 

1645 self.permission_denied(request, "Can not delete component") 

1646 instance.acting_user = request.user 

1647 component_removal.delay(instance.pk, request.user.pk) 

1648 return Response(status=HTTP_204_NO_CONTENT) 

1649 

1650 def add_link(self, request: Request, instance: Component): 

1651 if not request.user.has_perm("component.edit", instance): 1651 ↛ 1652line 1651 didn't jump to line 1652 because the condition on line 1651 was never true

1652 self.permission_denied(request, "Can not edit component") 

1653 if "project_slug" not in request.data: 1653 ↛ 1657line 1653 didn't jump to line 1657 because the condition on line 1653 was always true

1654 msg = "Missing 'project_slug' parameter" 

1655 raise ValidationError({"project_slug": msg}) 

1656 

1657 project_slug = request.data["project_slug"] 

1658 

1659 try: 

1660 project = request.user.allowed_projects.exclude(pk=instance.project_id).get( 

1661 slug=project_slug 

1662 ) 

1663 except Project.DoesNotExist as error: 

1664 msg = f"No project slug {project_slug!r} found!" 

1665 raise ValidationError({"project_slug": msg}) from error 

1666 

1667 instance.links.add(project) 

1668 serializer = self.serializer_class(instance, context={"request": request}) 

1669 

1670 return Response(data={"data": serializer.data}, status=HTTP_201_CREATED) 

1671 

1672 @extend_schema( 

1673 description="Return projects linked with a component.", methods=["get"] 

1674 ) 

1675 @extend_schema(description="Associate project with a component.", methods=["post"]) 

1676 @action(detail=True, methods=["get", "post"]) 

1677 def links(self, request: Request, **kwargs): 

1678 instance = self.get_object() 

1679 if request.method == "POST": 

1680 return self.add_link(request, instance) 

1681 

1682 queryset = instance.links.order_by("id") 

1683 page = self.paginate_queryset(queryset) 

1684 

1685 serializer = ProjectSerializer(page, many=True, context={"request": request}) 

1686 

1687 return self.get_paginated_response(serializer.data) 

1688 

1689 @extend_schema( 

1690 description="Remove association of a project with a component.", 

1691 methods=["delete"], 

1692 parameters=[OpenApiParameter("project_slug", str, OpenApiParameter.PATH)], 

1693 ) 

1694 @action(detail=True, methods=["delete"], url_path="links/(?P<project_slug>[^/.]+)") 

1695 def delete_links(self, request: Request, project__slug, slug, project_slug): 

1696 instance = self.get_object() 

1697 if not request.user.has_perm("component.edit", instance): 1697 ↛ 1698line 1697 didn't jump to line 1698 because the condition on line 1697 was never true

1698 self.permission_denied(request, "Can not edit component") 

1699 

1700 try: 

1701 project = instance.links.get(slug=project_slug) 

1702 except Project.DoesNotExist as error: 

1703 msg = "Project not found" 

1704 raise Http404(msg) from error 

1705 instance.links.remove(project) 

1706 return Response(status=HTTP_204_NO_CONTENT) 

1707 

1708 @action(detail=True, methods=["get"]) 

1709 def file(self, request: Request, **kwargs): 

1710 # Implementation is analogous to files#download_component, but we can't reuse 

1711 # that here because the lookup for the component is different 

1712 instance = self.get_object() 

1713 if not request.user.has_perm("translation.download", instance): 1713 ↛ 1714line 1713 didn't jump to line 1714 because the condition on line 1713 was never true

1714 self.permission_denied( 

1715 request, "Can not download all translations for the component" 

1716 ) 

1717 

1718 requested_format = request.query_params.get("format", "zip") 

1719 return download_multi( 

1720 cast("AuthenticatedHttpRequest", request), 

1721 instance.translation_set.prefetch_meta(), 

1722 [instance], 

1723 requested_format, 

1724 name=instance.full_slug.replace("/", "-"), 

1725 ) 

1726 

1727 

1728@extend_schema_view( 

1729 list=extend_schema(description="Return a list of memory results."), 

1730) 

1731class MemoryViewSet(viewsets.ModelViewSet, DestroyModelMixin): 

1732 """Memory API.""" 

1733 

1734 queryset = Memory.objects.none() 

1735 serializer_class = MemorySerializer 

1736 

1737 def get_queryset(self): 

1738 if not self.request.user.is_superuser: 1738 ↛ 1739line 1738 didn't jump to line 1739 because the condition on line 1738 was never true

1739 self.permission_denied(self.request, "Access not allowed") 

1740 # Use default database connection and not memory_db one (in case 

1741 # a custom router is used). 

1742 return Memory.objects.using("default").order_by("id") 

1743 

1744 def perm_check(self, request: Request, instance) -> None: 

1745 if not request.user.has_perm("memory.delete", instance): 1745 ↛ 1746line 1745 didn't jump to line 1746 because the condition on line 1745 was never true

1746 self.permission_denied(request, "Can not delete memory entry") 

1747 

1748 def destroy(self, request: Request, *args, **kwargs): 

1749 """Delete a memory object.""" 

1750 instance = self.get_object() 

1751 self.perm_check(request, instance) 

1752 return super().destroy(request, *args, **kwargs) 

1753 

1754 

1755@extend_schema_view( 

1756 list=extend_schema(description="Return a list of translations."), 

1757 retrieve=extend_schema(description="Return information about a translation."), 

1758) 

1759class TranslationViewSet(MultipleFieldViewSet, DestroyModelMixin): 

1760 """Translation components API.""" 

1761 

1762 queryset = Translation.objects.none() 

1763 serializer_class = TranslationSerializer 

1764 lookup_fields = ("component__project__slug", "component__slug", "language__code") 

1765 raw_urls = ("translation-file",) 

1766 

1767 def get_queryset(self): 

1768 return ( 

1769 Translation.objects.filter_access(self.request.user) 

1770 .prefetch(defer_huge=False) 

1771 .prefetch_related("component__source_language") 

1772 .prefetch_related("component__addon_set") 

1773 .prefetch_plurals() 

1774 .order_by("id") 

1775 ) 

1776 

1777 def paginate_queryset(self, queryset): 

1778 result = super().paginate_queryset(queryset) 

1779 if isinstance(queryset, TranslationQuerySet): 

1780 result = prefetch_stats(result) 

1781 processed: set[int] = set() 

1782 for translation in result: 

1783 project = translation.component.project 

1784 if project.id in processed: 

1785 continue 

1786 # Prefetch workflow settings 

1787 project.project_languages.preload_workflow_settings() 

1788 processed.add(project.id) 

1789 return result 

1790 

1791 @extend_schema(description="Upload new file with translations.", methods=["post"]) 

1792 @action( 

1793 detail=True, 

1794 methods=["get", "put", "post"], 

1795 parser_classes=( 

1796 parsers.MultiPartParser, 

1797 parsers.FormParser, 

1798 parsers.FileUploadParser, 

1799 ), 

1800 serializer_class=UploadRequestSerializer, 

1801 ) 

1802 def file(self, request: Request, **kwargs): 

1803 obj = self.get_object() 

1804 user = request.user 

1805 if request.method == "GET": 

1806 if obj.get_filename() is None: 

1807 msg = "No translation file!" 

1808 raise Http404(msg) 

1809 if not user.has_perm("translation.download", obj): 1809 ↛ 1810line 1809 didn't jump to line 1810 because the condition on line 1809 was never true

1810 raise PermissionDenied 

1811 fmt = self.format_kwarg or request.query_params.get("format") 

1812 query_string = request.GET.get("q", "") 

1813 if query_string and not fmt: 1813 ↛ 1814line 1813 didn't jump to line 1814 because the condition on line 1813 was never true

1814 raise ValidationError({"q": "Query string is ignored without format"}) 

1815 try: 

1816 parse_query(query_string) 

1817 except Exception as error: 

1818 raise ValidationError( 

1819 {"q": f"Could not parse query string: {error}"} 

1820 ) from error 

1821 try: 

1822 return download_translation_file(request, obj, fmt, query_string) 

1823 except Http404 as error: 

1824 raise ValidationError({"format": str(error)}) from error 

1825 

1826 if not (can_upload := user.has_perm("upload.perform", obj)): 

1827 self.permission_denied( 

1828 request, 

1829 can_upload.reason 

1830 if isinstance(can_upload, PermissionResult) 

1831 else "Insufficient privileges for uploading.", 

1832 ) 

1833 

1834 serializer = UploadRequestSerializer(data=request.data) 

1835 serializer.is_valid(raise_exception=True) 

1836 serializer.check_perms(request.user, obj) 

1837 

1838 data = serializer.validated_data 

1839 

1840 if obj.get_filename() is None and data["method"] != "source": 

1841 raise ValidationError( 

1842 {"method": "No translation file, try using method=source."} 

1843 ) 

1844 

1845 author_name = None 

1846 author_email = None 

1847 if request.user.has_perm("upload.authorship", obj): 

1848 author_name = data.get("author_name") 

1849 author_email = data.get("author_email") 

1850 

1851 try: 

1852 not_found, skipped, accepted, total = obj.handle_upload( 

1853 request, 

1854 data["file"], 

1855 data["conflicts"], 

1856 author_name, 

1857 author_email, 

1858 data["method"], 

1859 data["fuzzy"], 

1860 ) 

1861 except Exception as error: 

1862 report_error("Upload error", print_tb=True, project=obj.component.project) 

1863 raise ValidationError({"file": str(error)}) from error 

1864 

1865 return Response( 

1866 data={ 

1867 "not_found": not_found, 

1868 "skipped": skipped, 

1869 "accepted": accepted, 

1870 "total": total, 

1871 # Compatibility with older less detailed API 

1872 "result": accepted > 0, 

1873 "count": total, 

1874 } 

1875 ) 

1876 

1877 @extend_schema( 

1878 description="Return detailed translation statistics.", 

1879 methods=["get"], 

1880 tags=["translations", "statistics"], 

1881 ) 

1882 @action( 

1883 detail=True, 

1884 methods=["get"], 

1885 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

1886 ) 

1887 def statistics(self, request: Request, **kwargs): 

1888 obj = self.get_object() 

1889 

1890 serializer = StatisticsSerializer(obj, context={"request": request}) 

1891 

1892 return Response(serializer.data) 

1893 

1894 @extend_schema(description="Return a list of translation changes.", methods=["get"]) 

1895 @action(detail=True, methods=["get"]) 

1896 def changes(self, request: Request, **kwargs): 

1897 obj = self.get_object() 

1898 

1899 queryset = obj.change_set.prefetch().order() 

1900 queryset = ChangesFilterBackend().filter_queryset(request, queryset, self) 

1901 page = self.paginate_queryset(queryset) 

1902 

1903 serializer = ChangeSerializer(page, many=True, context={"request": request}) 

1904 

1905 return self.get_paginated_response(serializer.data) 

1906 

1907 @extend_schema(description="Return a list of translation units.", methods=["get"]) 

1908 @extend_schema(description="Add a new unit.", methods=["post"]) 

1909 @action(detail=True, methods=["get", "post"]) 

1910 def units(self, request: Request, **kwargs): 

1911 obj = self.get_object() 

1912 

1913 serializer_class: type[NewUnitSerializer] 

1914 if obj.component.template: 1914 ↛ 1915line 1914 didn't jump to line 1915 because the condition on line 1914 was never true

1915 serializer_class = MonolingualUnitSerializer 

1916 elif obj.is_source: 

1917 serializer_class = BilingualSourceUnitSerializer 

1918 else: 

1919 serializer_class = BilingualUnitSerializer 

1920 

1921 if request.method == "POST": 

1922 with transaction.atomic(): 

1923 if not (can_add := request.user.has_perm("unit.add", obj)): 

1924 self.permission_denied(request, can_add.reason) 

1925 serializer = serializer_class( 

1926 data=request.data, context={"translation": obj} 

1927 ) 

1928 serializer.is_valid(raise_exception=True) 

1929 

1930 unit = obj.add_unit(request, **serializer.as_kwargs()) 

1931 outserializer = UnitSerializer(unit, context={"request": request}) 

1932 return Response(outserializer.data, status=HTTP_200_OK) 

1933 

1934 query_string = request.GET.get("q", "") 

1935 try: 

1936 parse_query(query_string) 

1937 except Exception as error: 

1938 msg = f"Could not parse query string: {error}" 

1939 raise ValidationError({"q": msg}) from error 

1940 

1941 queryset = obj.unit_set.search(query_string).order_by("id").prefetch_full() 

1942 page = self.paginate_queryset(queryset) 

1943 

1944 serializer = UnitSerializer(page, many=True, context={"request": request}) 

1945 

1946 return self.get_paginated_response(serializer.data) 

1947 

1948 @extend_schema(description="Trigger automatic translation.", methods=["post"]) 

1949 @action(detail=True, methods=["post"]) 

1950 def autotranslate(self, request: Request, **kwargs): 

1951 translation = self.get_object() 

1952 if not request.user.has_perm("translation.auto", translation): 

1953 self.permission_denied(request, "Can not auto translate") 

1954 if translation.component.locked: 1954 ↛ 1955line 1954 didn't jump to line 1955 because the condition on line 1954 was never true

1955 self.permission_denied(request, "Component is locked") 

1956 

1957 autoform = AutoForm(translation.component, request.user, request.data) 

1958 if not autoform.is_valid(): 1958 ↛ 1968line 1958 didn't jump to line 1968 because the condition on line 1958 was always true

1959 errors: dict[str, str] = {} 

1960 for field in autoform: 

1961 for error in field.errors: 

1962 if field.name in errors: 1962 ↛ 1963line 1962 didn't jump to line 1963 because the condition on line 1962 was never true

1963 errors[field.name] += f", {error}" 

1964 else: 

1965 errors[field.name] = str(error) 

1966 raise ValidationError(errors) 

1967 

1968 auto = AutoTranslate( 

1969 user=request.user, 

1970 translation=translation, 

1971 q=autoform.cleaned_data["q"], 

1972 mode=autoform.cleaned_data["mode"], 

1973 ) 

1974 message = auto.perform( 

1975 auto_source=autoform.cleaned_data["auto_source"], 

1976 source=autoform.cleaned_data["component"], 

1977 engines=autoform.cleaned_data["engines"], 

1978 threshold=autoform.cleaned_data["threshold"], 

1979 ) 

1980 

1981 return Response( 

1982 data={"details": message}, 

1983 status=HTTP_200_OK, 

1984 ) 

1985 

1986 def destroy(self, request: Request, *args, **kwargs): 

1987 """Delete a translation.""" 

1988 instance = self.get_object() 

1989 if not request.user.has_perm("translation.delete", instance): 

1990 self.permission_denied(request, "Can not delete translation") 

1991 instance.remove(request.user) 

1992 return Response(status=HTTP_204_NO_CONTENT) 

1993 

1994 

1995@extend_schema_view( 

1996 list=extend_schema( 

1997 description="Return a list of all languages the user has access to." 

1998 ), 

1999 retrieve=extend_schema(description="Return information about a language."), 

2000 partial_update=extend_schema(description="Change the language parameters."), 

2001) 

2002class LanguageViewSet(viewsets.ModelViewSet): 

2003 """Languages API.""" 

2004 

2005 queryset = Language.objects.none() 

2006 serializer_class = LanguageSerializer 

2007 lookup_field = "code" 

2008 

2009 def get_queryset(self): 

2010 if self.request.user.has_perm("language.edit"): 2010 ↛ 2012line 2010 didn't jump to line 2012 because the condition on line 2010 was always true

2011 return Language.objects.order_by("id").prefetch() 

2012 return Language.objects.have_translation().order_by("id").prefetch() 

2013 

2014 def perm_check(self, request: Request) -> None: 

2015 if not request.user.has_perm("language.edit"): 2015 ↛ 2016line 2015 didn't jump to line 2016 because the condition on line 2015 was never true

2016 self.permission_denied(request, "Can not manage languages") 

2017 

2018 def update(self, request: Request, *args, **kwargs): 

2019 """Change the language parameters.""" 

2020 self.perm_check(request) 

2021 return super().update(request, *args, **kwargs) 

2022 

2023 def create(self, request: Request, *args, **kwargs): 

2024 """Create a new language.""" 

2025 self.perm_check(request) 

2026 return super().create(request, *args, **kwargs) 

2027 

2028 def destroy(self, request: Request, *args, **kwargs): 

2029 """Delete the language.""" 

2030 self.perm_check(request) 

2031 return super().destroy(request, *args, **kwargs) 

2032 

2033 @extend_schema( 

2034 description="Return statistics for a language.", 

2035 methods=["get"], 

2036 tags=["languages", "statistics"], 

2037 ) 

2038 @action( 

2039 detail=True, 

2040 methods=["get"], 

2041 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

2042 ) 

2043 def statistics(self, request: Request, **kwargs): 

2044 obj = self.get_object() 

2045 

2046 serializer = StatisticsSerializer(obj, context={"request": request}) 

2047 

2048 return Response(serializer.data) 

2049 

2050 

2051@extend_schema_view( 

2052 list=extend_schema(description="Return a list of translation units."), 

2053 retrieve=extend_schema(description="Return information about translation unit."), 

2054 update=extend_schema(description="Perform full update on translation unit."), 

2055 partial_update=extend_schema( 

2056 description="Perform partial update on translation unit." 

2057 ), 

2058) 

2059class UnitViewSet(viewsets.ReadOnlyModelViewSet, UpdateModelMixin, DestroyModelMixin): 

2060 """Units API.""" 

2061 

2062 pagination_class = LargePagination 

2063 

2064 queryset = Unit.objects.none() 

2065 

2066 def get_serializer_class(self): 

2067 """Get correct serializer based on action.""" 

2068 if self.action in {"list", "retrieve"}: 

2069 return UnitSerializer 

2070 return UnitWriteSerializer 

2071 

2072 def get_queryset(self): 

2073 return ( 

2074 Unit.objects.filter_access(self.request.user) 

2075 .prefetch() 

2076 .prefetch_full() 

2077 .order_by("id") 

2078 ) 

2079 

2080 def filter_queryset(self, queryset): 

2081 result = super().filter_queryset(queryset) 

2082 query_string = self.request.GET.get("q", "") 

2083 try: 

2084 parse_query(query_string) 

2085 except Exception as error: 

2086 msg = f"Could not parse query string: {error}" 

2087 raise ValidationError({"q": msg}) from error 

2088 if query_string: 2088 ↛ 2089line 2088 didn't jump to line 2089 because the condition on line 2088 was never true

2089 result = result.search(query_string) 

2090 return result 

2091 

2092 @transaction.atomic 

2093 def perform_update(self, serializer) -> None: # noqa: C901 

2094 data = serializer.validated_data 

2095 do_translate = "target" in data or "state" in data 

2096 do_source = "extra_flags" in data or "explanation" in data or "labels" in data 

2097 unit = serializer.instance 

2098 translation = unit.translation 

2099 request = self.request 

2100 user = request.user 

2101 

2102 new_target = data.get("target", []) 

2103 new_state = data.get("state", None) 

2104 

2105 # Sanity and permission checks 

2106 if do_source and ( 

2107 not unit.is_source or not user.has_perm("source.edit", translation) 

2108 ): 

2109 self.permission_denied( 

2110 request, "Source strings properties can be set only on source strings" 

2111 ) 

2112 

2113 if do_translate: 

2114 new_target_copy = new_target[:] 

2115 if new_target_copy != unit.adjust_plurals(new_target): 

2116 raise ValidationError({"target": "Number of plurals does not match"}) 

2117 

2118 if unit.readonly: 

2119 self.permission_denied(request, "The string is read-only.") 

2120 if not new_target or new_state is None: 2120 ↛ 2124line 2120 didn't jump to line 2124 because the condition on line 2120 was always true

2121 msg = "Please provide both state and target for a partial update" 

2122 raise ValidationError({"state": msg, "target": msg}) 

2123 

2124 if new_state not in { 

2125 STATE_APPROVED, 

2126 STATE_TRANSLATED, 

2127 STATE_FUZZY, 

2128 STATE_EMPTY, 

2129 }: 

2130 raise ValidationError({"state": "Invalid state"}) 

2131 

2132 if new_state == STATE_EMPTY and any(new_target): 

2133 raise ValidationError( 

2134 {"state": "Can not use empty state with non empty target"} 

2135 ) 

2136 

2137 if new_state != STATE_EMPTY and not any(new_target): 

2138 raise ValidationError( 

2139 {"state": "Can not use non empty state with empty target"} 

2140 ) 

2141 

2142 can_edit = user.has_perm("unit.edit", unit) 

2143 if not can_edit: 

2144 self.permission_denied(request, can_edit.reason) 

2145 

2146 if new_state == STATE_APPROVED: 

2147 can_review = user.has_perm("unit.review", translation) 

2148 if not can_review: 

2149 raise ValidationError({"state": can_review.reason}) 

2150 

2151 # Update attributes 

2152 if do_source: 

2153 fields = ["extra_flags", "explanation"] 

2154 for name in fields: 

2155 try: 

2156 setattr(unit, name, data[name]) 

2157 except KeyError: 

2158 continue 

2159 if "labels" in data: 

2160 unit.labels.set(data["labels"]) 

2161 unit.save(update_fields=fields) 

2162 

2163 # Handle translate 

2164 if do_translate: 2164 ↛ 2165line 2164 didn't jump to line 2165 because the condition on line 2164 was never true

2165 unit.translate(user, new_target, new_state) 

2166 

2167 def destroy(self, request: Request, *args, **kwargs): 

2168 """Delete a translation unit.""" 

2169 obj = self.get_object() 

2170 can_delete = request.user.has_perm("unit.delete", obj) 

2171 if not can_delete: 2171 ↛ 2173line 2171 didn't jump to line 2173 because the condition on line 2171 was always true

2172 self.permission_denied(request, can_delete.reason) 

2173 try: 

2174 obj.translation.delete_unit(request, obj) 

2175 except FileParseError as error: 

2176 obj.translation.component.update_import_alerts(delete=False) 

2177 return Response( 

2178 data={"error": f"Could not remove the string: {error}"}, 

2179 status=HTTP_500_INTERNAL_SERVER_ERROR, 

2180 ) 

2181 return Response(status=HTTP_204_NO_CONTENT) 

2182 

2183 @action(detail=True, methods=["get"]) 

2184 def translations(self, request: Request, *args, **kwargs): 

2185 unit = self.get_object() 

2186 user = request.user 

2187 user.check_access_component(unit.translation.component) 

2188 

2189 if not unit.is_source: 

2190 raise NotSourceUnit 

2191 

2192 translation_units = ( 

2193 unit.source_unit.unit_set.exclude(pk=unit.pk).prefetch().prefetch_full() 

2194 ) 

2195 serializer = UnitSerializer( 

2196 translation_units, many=True, context={"request": request} 

2197 ) 

2198 return Response(serializer.data) 

2199 

2200 @extend_schema( 

2201 description="Add a comment to the unit.", 

2202 methods=["post"], 

2203 request=CommentSerializer, 

2204 ) 

2205 @action(detail=True, methods=["post"], serializer_class=CommentSerializer) 

2206 def comments(self, request, *args, **kwargs): 

2207 """Add a new comment to a unit.""" 

2208 unit = self.get_object() 

2209 user = request.user 

2210 

2211 if not user.has_perm("comment.add", unit.translation): 2211 ↛ 2212line 2211 didn't jump to line 2212 because the condition on line 2211 was never true

2212 self.permission_denied(request) 

2213 

2214 serializer = CommentSerializer( 

2215 data=request.data, 

2216 context={ 

2217 "unit": unit, 

2218 "request": request, 

2219 }, 

2220 ) 

2221 serializer.is_valid(raise_exception=True) 

2222 

2223 timestamp = serializer.validated_data.get("timestamp") 

2224 user_email = serializer.validated_data.get("user_email") 

2225 if (timestamp is not None or user_email is not None) and ( 2225 ↛ 2228line 2225 didn't jump to line 2228 because the condition on line 2225 was never true

2226 not user.has_perm("project.edit", unit.translation.component.project) 

2227 ): 

2228 self.permission_denied(request) 

2229 

2230 serializer.save() 

2231 return Response(serializer.data, status=HTTP_201_CREATED) 

2232 

2233 

2234@extend_schema_view( 

2235 list=extend_schema(description="Return a list of screenshot string information."), 

2236 retrieve=extend_schema( 

2237 description="Return information about screenshot information." 

2238 ), 

2239 partial_update=extend_schema( 

2240 description="Edit partial information about screenshot." 

2241 ), 

2242) 

2243class ScreenshotViewSet(DownloadViewSet, viewsets.ModelViewSet): 

2244 """Screenshots API.""" 

2245 

2246 queryset = Screenshot.objects.none() 

2247 serializer_class = ScreenshotSerializer 

2248 raw_urls = ("screenshot-file",) 

2249 raw_formats = () 

2250 

2251 def get_queryset(self): 

2252 return Screenshot.objects.filter_access(self.request.user).order_by("id") 

2253 

2254 @extend_schema(description="Download the screenshot image.", methods=["get"]) 

2255 @extend_schema(description="Replace screenshot image.", methods=["post"]) 

2256 @action( 

2257 detail=True, 

2258 methods=["get", "put", "post"], 

2259 parser_classes=( 

2260 parsers.MultiPartParser, 

2261 parsers.FormParser, 

2262 parsers.FileUploadParser, 

2263 ), 

2264 serializer_class=ScreenshotFileSerializer, 

2265 ) 

2266 def file(self, request: Request, **kwargs): 

2267 obj = self.get_object() 

2268 if request.method == "GET": 

2269 return self.download_file(obj.image.path, "application/binary") 

2270 

2271 if not request.user.has_perm("screenshot.edit", obj.translation): 

2272 raise PermissionDenied 

2273 

2274 serializer = ScreenshotFileSerializer(data=request.data) 

2275 serializer.is_valid(raise_exception=True) 

2276 

2277 obj.image.save( 

2278 serializer.validated_data["image"].name, serializer.validated_data["image"] 

2279 ) 

2280 

2281 return Response(data={"result": True}) 

2282 

2283 @extend_schema( 

2284 description="Associate source string with screenshot.", methods=["post"] 

2285 ) 

2286 @action(detail=True, methods=["post"]) 

2287 @transaction.atomic 

2288 def units(self, request: Request, **kwargs): 

2289 obj = self.get_object() 

2290 

2291 if not request.user.has_perm("screenshot.edit", obj.translation): 

2292 raise PermissionDenied 

2293 

2294 if "unit_id" not in request.data: 

2295 raise ValidationError({"unit_id": "This field is required."}) 

2296 

2297 try: 

2298 unit = obj.translation.unit_set.get(pk=int(request.data["unit_id"])) 

2299 except (Unit.DoesNotExist, ValueError) as error: 

2300 raise ValidationError({"unit_id": str(error)}) from error 

2301 

2302 obj.units.add(unit) 

2303 serializer = ScreenshotSerializer(obj, context={"request": request}) 

2304 

2305 return Response(serializer.data, status=HTTP_200_OK) 

2306 

2307 @extend_schema( 

2308 description="Remove source string association with screenshot.", 

2309 methods=["delete"], 

2310 ) 

2311 @action(detail=True, methods=["delete"], url_path="units/(?P<unit_id>[0-9]+)") 

2312 def delete_units(self, request: Request, pk, unit_id): 

2313 obj = self.get_object() 

2314 if not request.user.has_perm("screenshot.edit", obj.translation): 

2315 raise PermissionDenied 

2316 

2317 try: 

2318 unit = obj.translation.unit_set.get(pk=unit_id) 

2319 except Unit.DoesNotExist as error: 

2320 raise Http404(str(error)) from error 

2321 obj.units.remove(unit) 

2322 return Response(status=HTTP_204_NO_CONTENT) 

2323 

2324 def create(self, request: Request, *args, **kwargs): 

2325 """Create a new screenshot.""" 

2326 required_params = ["project_slug", "component_slug", "language_code"] 

2327 for param in required_params: 2327 ↛ 2331line 2327 didn't jump to line 2331 because the loop on line 2327 didn't complete

2328 if param not in request.data: 2328 ↛ 2327line 2328 didn't jump to line 2327 because the condition on line 2328 was always true

2329 raise ValidationError({param: "This field is required."}) 

2330 

2331 try: 

2332 translation = Translation.objects.get( 

2333 component__project__slug=request.data["project_slug"], 

2334 component__slug=request.data["component_slug"], 

2335 language__code=request.data["language_code"], 

2336 ) 

2337 except Translation.DoesNotExist as error: 

2338 raise ValidationError( 

2339 {key: str(error) for key in required_params} 

2340 ) from error 

2341 

2342 if not request.user.has_perm("screenshot.add", translation): 

2343 self.permission_denied(request, "Can not add screenshot.") 

2344 

2345 with transaction.atomic(): 

2346 serializer = ScreenshotCreateSerializer( 

2347 data=request.data, context={"request": request} 

2348 ) 

2349 serializer.is_valid(raise_exception=True) 

2350 instance = serializer.save(translation=translation, user=request.user) 

2351 instance.change_set.create( 

2352 action=ActionEvents.SCREENSHOT_ADDED, 

2353 user=request.user, 

2354 target=instance.name, 

2355 ) 

2356 return Response(serializer.data, status=HTTP_201_CREATED) 

2357 

2358 def update(self, request: Request, *args, **kwargs): 

2359 """Edit full information about screenshot.""" 

2360 instance = self.get_object() 

2361 if not request.user.has_perm("screenshot.edit", instance.translation): 

2362 self.permission_denied(request, "Can not edit screenshot.") 

2363 return super().update(request, *args, **kwargs) 

2364 

2365 def destroy(self, request: Request, *args, **kwargs): 

2366 """Delete screenshot.""" 

2367 instance = self.get_object() 

2368 if not request.user.has_perm("screenshot.delete", instance.translation): 

2369 self.permission_denied(request, "Can not delete screenshot.") 

2370 return super().destroy(request, *args, **kwargs) 

2371 

2372 

2373class ChangeFilter(filters.FilterSet): 

2374 timestamp = filters.IsoDateTimeFromToRangeFilter() 

2375 action = filters.MultipleChoiceFilter(choices=ActionEvents.choices) 

2376 user = filters.CharFilter(field_name="user__username") 

2377 

2378 class Meta: 

2379 model = Change 

2380 fields = ("action", "user", "timestamp") 

2381 

2382 

2383class ChangesFilterBackend(filters.DjangoFilterBackend): 

2384 def get_filterset_class(self, view, queryset=None): 

2385 return ChangeFilter 

2386 

2387 

2388@extend_schema_view( 

2389 list=extend_schema(description="Return a list of translation changes."), 

2390 retrieve=extend_schema( 

2391 description="Return information about a translation change." 

2392 ), 

2393) 

2394class ChangeViewSet(viewsets.ReadOnlyModelViewSet): 

2395 """Changes API.""" 

2396 

2397 queryset = Change.objects.none() 

2398 serializer_class = ChangeSerializer 

2399 filter_backends = (ChangesFilterBackend,) 

2400 

2401 def get_queryset(self): 

2402 return Change.objects.last_changes(self.request.user) 

2403 

2404 def paginate_queryset(self, queryset): 

2405 result = super().paginate_queryset(queryset) 

2406 return Change.objects.preload_list(result) 

2407 

2408 

2409@extend_schema_view( 

2410 list=extend_schema(description="Return a list of component lists."), 

2411 retrieve=extend_schema(description="Return information about component list."), 

2412 partial_update=extend_schema(description="Change the component list parameters."), 

2413) 

2414class ComponentListViewSet(viewsets.ModelViewSet): 

2415 """Component lists API.""" 

2416 

2417 queryset = ComponentList.objects.none() 

2418 serializer_class = ComponentListSerializer 

2419 lookup_field = "slug" 

2420 request: Request # type: ignore[assignment] 

2421 

2422 def get_queryset(self): 

2423 return ( 

2424 ComponentList.objects.filter( 

2425 Q(components__project__in=self.request.user.allowed_projects) 

2426 | Q(components__isnull=True) 

2427 ) 

2428 .prefetch_related("components__project", "autocomponentlist_set") 

2429 .order_by("id") 

2430 .distinct() 

2431 ) 

2432 

2433 def perm_check(self, request: Request) -> None: 

2434 if not request.user.has_perm("componentlist.edit"): 2434 ↛ 2435line 2434 didn't jump to line 2435 because the condition on line 2434 was never true

2435 self.permission_denied(request, "Can not manage component lists") 

2436 

2437 def update(self, request: Request, *args, **kwargs): 

2438 """Change the component list parameters.""" 

2439 self.perm_check(request) 

2440 return super().update(request, *args, **kwargs) 

2441 

2442 def create(self, request: Request, *args, **kwargs): 

2443 """Create a new component list.""" 

2444 self.perm_check(request) 

2445 return super().create(request, *args, **kwargs) 

2446 

2447 def destroy(self, request: Request, *args, **kwargs): 

2448 """Delete the component list.""" 

2449 self.perm_check(request) 

2450 return super().destroy(request, *args, **kwargs) 

2451 

2452 @extend_schema( 

2453 description="Associate component with a component list.", methods=["post"] 

2454 ) 

2455 @action(detail=True, methods=["post", "get"]) 

2456 def components(self, request: Request, **kwargs): 

2457 obj = self.get_object() 

2458 if request.method == "POST": 

2459 self.perm_check(request) 

2460 

2461 if "component_id" not in request.data: 2461 ↛ 2464line 2461 didn't jump to line 2464 because the condition on line 2461 was always true

2462 raise ValidationError({"component_id": "This field is required."}) 

2463 

2464 try: 

2465 component = Component.objects.filter_access(self.request.user).get( 

2466 pk=int(request.data["component_id"]), 

2467 ) 

2468 except (Component.DoesNotExist, ValueError) as error: 

2469 raise ValidationError({"component_id": str(error)}) from error 

2470 

2471 obj.components.add(component) 

2472 serializer = self.serializer_class(obj, context={"request": request}) 

2473 

2474 return Response(serializer.data, status=HTTP_200_OK) 

2475 

2476 queryset = ( 

2477 obj.components.filter_access(self.request.user).prefetch().order_by("id") 

2478 ) 

2479 page = self.paginate_queryset(queryset) 

2480 

2481 serializer = ComponentSerializer(page, many=True, context={"request": request}) 

2482 

2483 return self.get_paginated_response(serializer.data) 

2484 

2485 @extend_schema( 

2486 description="Disassociate a component from the component list.", 

2487 methods=["delete"], 

2488 parameters=[OpenApiParameter("component_slug", str, OpenApiParameter.PATH)], 

2489 ) 

2490 @action( 

2491 detail=True, 

2492 methods=["delete"], 

2493 url_path="components/(?P<component_slug>[^/.]+)", 

2494 ) 

2495 def delete_components(self, request: Request, slug, component_slug): 

2496 obj = self.get_object() 

2497 self.perm_check(request) 

2498 

2499 try: 

2500 component = obj.components.get(slug=component_slug) 

2501 except Component.DoesNotExist as error: 

2502 raise Http404(str(error)) from error 

2503 obj.components.remove(component) 

2504 return Response(status=HTTP_204_NO_CONTENT) 

2505 

2506 

2507@extend_schema_view(list=extend_schema(description="List available categories.")) 

2508class CategoryViewSet(viewsets.ModelViewSet): 

2509 """Category API.""" 

2510 

2511 queryset = Category.objects.none() 

2512 serializer_class = CategorySerializer 

2513 lookup_field = "pk" 

2514 request: Request # type: ignore[assignment] 

2515 

2516 def get_queryset(self): 

2517 return Category.objects.filter( 

2518 project__in=self.request.user.allowed_projects 

2519 ).order_by("id") 

2520 

2521 def perm_check(self, request: Request, instance) -> None: 

2522 if not request.user.has_perm("project.edit", instance): 2522 ↛ 2523line 2522 didn't jump to line 2523 because the condition on line 2522 was never true

2523 self.permission_denied(request, "Can not manage categories") 

2524 

2525 def update(self, request: Request, *args, **kwargs): 

2526 """Edit full information about category.""" 

2527 self.perm_check(request, self.get_object()) 

2528 return super().update(request, *args, **kwargs) 

2529 

2530 def destroy(self, request: Request, *args, **kwargs): 

2531 """Delete category.""" 

2532 instance = self.get_object() 

2533 self.perm_check(request, instance) 

2534 category_removal.delay(instance.pk, request.user.pk) 

2535 return Response(status=HTTP_204_NO_CONTENT) 

2536 

2537 def perform_create(self, serializer) -> None: 

2538 """Create a new category.""" 

2539 if not self.request.user.has_perm( 2539 ↛ 2542line 2539 didn't jump to line 2542 because the condition on line 2539 was never true

2540 "project.edit", serializer.validated_data["project"] 

2541 ): 

2542 self.permission_denied( 

2543 self.request, "Can not manage categories in this project" 

2544 ) 

2545 serializer.save() 

2546 

2547 def perform_update(self, serializer) -> None: 

2548 if not self.request.user.has_perm( 2548 ↛ 2552line 2548 didn't jump to line 2552 because the condition on line 2548 was never true

2549 "project.edit", 

2550 serializer.validated_data.get("project", serializer.instance.project), 

2551 ): 

2552 self.permission_denied( 

2553 self.request, "Can not manage categories in this project" 

2554 ) 

2555 serializer.instance.acting_user = self.request.user 

2556 serializer.save() 

2557 

2558 @extend_schema( 

2559 description="""Return statistics for a category.""", 

2560 methods=["get"], 

2561 tags=["categories", "statistics"], 

2562 ) 

2563 @action( 

2564 detail=True, 

2565 methods=["get"], 

2566 renderer_classes=(*api_settings.DEFAULT_RENDERER_CLASSES, FlatJsonRenderer), 

2567 ) 

2568 def statistics(self, request: Request, **kwargs): 

2569 obj = self.get_object() 

2570 

2571 serializer = StatisticsSerializer(obj, context={"request": request}) 

2572 

2573 return Response(serializer.data) 

2574 

2575 

2576class Metrics(APIView): 

2577 """Metrics view for monitoring.""" 

2578 

2579 permission_classes = (IsAuthenticated,) 

2580 renderer_classes = (*api_settings.DEFAULT_RENDERER_CLASSES, OpenMetricsRenderer) 

2581 serializer_class = MetricsSerializer 

2582 

2583 def get(self, request: Request, format=None): # noqa: A002 

2584 """Return server metrics.""" 

2585 stats = GlobalStats() 

2586 serializer = self.serializer_class(stats) 

2587 return Response(serializer.data) 

2588 

2589 

2590class Search(APIView): 

2591 """Site-wide search endpoint.""" 

2592 

2593 serializer_class = None 

2594 

2595 def get(self, request: Request, format=None): # noqa: A002 

2596 """Return site-wide search results as a list.""" 

2597 user = request.user 

2598 projects = user.allowed_projects 

2599 components = Component.objects.filter(project__in=projects) 

2600 category = Category.objects.filter(project__in=projects) 

2601 results: list[dict[str, str]] = [] 

2602 query = request.GET.get("q") 

2603 if query and "\x00" not in query: 2603 ↛ 2604line 2603 didn't jump to line 2604 because the condition on line 2603 was never true

2604 results.extend( 

2605 { 

2606 "url": project.get_absolute_url(), 

2607 "name": project.name, 

2608 "category": gettext("Project"), 

2609 } 

2610 for project in projects.search(query).order()[:5] 

2611 ) 

2612 results.extend( 

2613 { 

2614 "url": category.get_absolute_url(), 

2615 "name": str(category), 

2616 "category": gettext("Category"), 

2617 } 

2618 for category in category.search(query).order()[:5] 

2619 ) 

2620 results.extend( 

2621 { 

2622 "url": component.get_absolute_url(), 

2623 "name": str(component), 

2624 "category": gettext("Component"), 

2625 } 

2626 for component in components.search(query).order()[:5] 

2627 ) 

2628 results.extend( 

2629 { 

2630 "url": user.get_absolute_url(), 

2631 "name": user.username, 

2632 "category": gettext("User"), 

2633 } 

2634 for user in User.objects.search(query, parser="plain").order()[:5] 

2635 ) 

2636 results.extend( 

2637 { 

2638 "url": language.get_absolute_url(), 

2639 "name": language.name, 

2640 "category": gettext("Language"), 

2641 } 

2642 for language in Language.objects.search(query).order()[:5] 

2643 ) 

2644 

2645 return Response(results) 

2646 

2647 

2648class TasksViewSet(ViewSet): 

2649 # Task-related data is handled and queried to Celery. 

2650 # There is no Django model associated with tasks. 

2651 serializer_class = None 

2652 

2653 def get_task( 

2654 self, request, pk, permission: str | None = None 

2655 ) -> tuple[AsyncResult, Component | None]: 

2656 obj: Model 

2657 component: Component 

2658 task = AsyncResult(str(pk)) 

2659 result = task.result 

2660 if task.state == "PENDING" or isinstance(result, Exception): 2660 ↛ 2663line 2660 didn't jump to line 2663 because the condition on line 2660 was always true

2661 component = None 

2662 else: 

2663 if result is None: 

2664 msg = "Task not found" 

2665 raise Http404(msg) 

2666 

2667 # Extract related object for permission check 

2668 if "translation" in result: 

2669 obj = get_object_or_404(Translation, pk=result["translation"]) 

2670 component = obj.component 

2671 elif "component" in result: 

2672 component = obj = get_object_or_404(Component, pk=result["component"]) 

2673 else: 

2674 msg = "Invalid task" 

2675 raise Http404(msg) 

2676 

2677 # Check access or permission 

2678 if permission: 

2679 if not request.user.has_perm(permission, obj): 

2680 raise PermissionDenied 

2681 elif not request.user.can_access_component(component): 

2682 raise PermissionDenied 

2683 

2684 return task, component 

2685 

2686 @extend_schema(description="Return information about a task", methods=["get"]) 

2687 def retrieve(self, request: Request, pk=None): 

2688 task, _component = self.get_task(request, pk) 

2689 result = task.result 

2690 return Response( 

2691 { 

2692 "completed": task.ready(), 

2693 "progress": get_task_progress(task), 

2694 "result": str(result) if isinstance(result, Exception) else result, 

2695 "log": "\n".join(cache.get(f"task-log-{task.id}", [])), 

2696 } 

2697 ) 

2698 

2699 def destroy(self, request: Request, pk=None): 

2700 task, component = self.get_task(request, pk, "component.edit") 

2701 if not task.ready() and component is not None: 2701 ↛ 2702line 2701 didn't jump to line 2702 because the condition on line 2701 was never true

2702 task.revoke(terminate=True) 

2703 # Unlink task from component 

2704 if component.background_task_id == pk: 

2705 component.delete_background_task() 

2706 return Response(status=HTTP_204_NO_CONTENT) 

2707 

2708 

2709@extend_schema_view( 

2710 list=extend_schema(description="Return a list of add-ons."), 

2711 retrieve=extend_schema(description="Returns information about add-on information."), 

2712 partial_update=extend_schema(description="Edit partial information about add-on."), 

2713) 

2714class AddonViewSet(viewsets.ReadOnlyModelViewSet, UpdateModelMixin, DestroyModelMixin): 

2715 queryset = Addon.objects.all() 

2716 serializer_class = AddonSerializer 

2717 

2718 def perm_check(self, request: Request, instance: Addon) -> None: 

2719 if instance.component and not request.user.has_perm( 

2720 "component.edit", instance.component 

2721 ): 

2722 self.permission_denied(request, "Can not manage addons") 

2723 if instance.project and not request.user.has_perm( 

2724 "project.edit", instance.project 

2725 ): 

2726 self.permission_denied(request, "Can not manage addons") 

2727 

2728 def update(self, request: Request, *args, **kwargs): 

2729 """Edit full information about add-on.""" 

2730 instance = self.get_object() 

2731 if instance.component: 

2732 instance.component.acting_user = request.user 

2733 if instance.project: 

2734 instance.project.acting_user = request.user 

2735 self.perm_check(request, instance) 

2736 return super().update(request, *args, **kwargs) 

2737 

2738 def destroy(self, request: Request, *args, **kwargs): 

2739 """Delete add-on.""" 

2740 instance = self.get_object() 

2741 if instance.component: 

2742 instance.component.acting_user = request.user 

2743 if instance.project: 

2744 instance.project.acting_user = request.user 

2745 self.perm_check(request, instance) 

2746 return super().destroy(request, *args, **kwargs)