Conversation
Contributor
Reviewer's GuideThis PR adds token-based authentication to the calendar event API by extending serializers and services to handle management tokens alongside regular user authentication, defines new public endpoints in the OpenAPI schema, and provides a dedicated token-driven viewset with comprehensive integration tests. Sequence diagram for token-based event creation via public endpointsequenceDiagram
actor Client
participant API["/public/organizations/{organization_id}/events/"]
participant ViewSet["TokenCalendarEventViewSet"]
participant CalendarService
participant CalendarPermissionService
participant DB["CalendarEvent DB"]
Client->>API: POST /public/organizations/{organization_id}/events/ (Bearer token)
API->>ViewSet: create(request)
ViewSet->>CalendarService: initialize_without_provider(token, organization)
CalendarService->>CalendarPermissionService: initialize_with_token(token, organization_id)
CalendarPermissionService-->>CalendarService: permission validated
ViewSet->>DB: create CalendarEvent
DB-->>ViewSet: CalendarEvent created
ViewSet-->>API: return created event
API-->>Client: 201 Created
Sequence diagram for token-based event update via public endpointsequenceDiagram
actor Client
participant API["/public/organizations/{organization_id}/events/{id}/"]
participant ViewSet["TokenCalendarEventViewSet"]
participant CalendarService
participant CalendarPermissionService
participant DB["CalendarEvent DB"]
Client->>API: PUT/PATCH /public/organizations/{organization_id}/events/{id}/ (Bearer token)
API->>ViewSet: update/partial_update(request)
ViewSet->>CalendarService: initialize_without_provider(token, organization)
CalendarService->>CalendarPermissionService: initialize_with_token(token, organization_id)
CalendarPermissionService-->>CalendarService: permission validated
ViewSet->>DB: update CalendarEvent
DB-->>ViewSet: CalendarEvent updated
ViewSet-->>API: return updated event
API-->>Client: 200 OK
Class diagram for token-based calendar event viewset and mixinclassDiagram
class TokenAuthenticationMixin {
+calendar_service: CalendarService | None
+extract_token_from_header(request: Request) tuple[str, int]
+get_calendar_service(request: Request, organization) CalendarService
}
class TokenCalendarEventViewSet {
+serializer_class: CalendarEventSerializer
+queryset: CalendarEvent.objects.all()
+authentication_classes: tuple
+permission_classes: tuple
+get_serializer_context()
+get_queryset()
+create()
+update()
+partial_update()
+destroy()
+list()
+retrieve()
+perform_create(serializer)
+perform_update(serializer)
+perform_destroy(instance)
}
TokenCalendarEventViewSet --|> TokenAuthenticationMixin
TokenCalendarEventViewSet --|> NoListVintaScheduleModelViewSet
TokenCalendarEventViewSet o-- CalendarEventSerializer
TokenCalendarEventViewSet o-- CalendarService
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
Blocking issues:
- Detected a Generic API Key, potentially exposing access to various services and sensitive operations. (link)
General comments:
- The CalendarEventSerializer init has grown very large and mixes multiple responsibilities—consider extracting token/user/org context setup and queryset configuration into smaller helper methods or a dedicated context manager for readability and maintainability.
- The TokenCalendarEventViewSet repeats the same token extraction and permission check logic in each action—extract this into a decorator or shared base method to DRY up the code and reduce boilerplate.
- The WriteOnlyVintaScheduleModelViewSet class description says it doesn’t allow updates, yet it includes UpdateModelMixin—either update the documentation or remove the conflicting mixin to align behavior and docs.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The CalendarEventSerializer __init__ has grown very large and mixes multiple responsibilities—consider extracting token/user/org context setup and queryset configuration into smaller helper methods or a dedicated context manager for readability and maintainability.
- The TokenCalendarEventViewSet repeats the same token extraction and permission check logic in each action—extract this into a decorator or shared base method to DRY up the code and reduce boilerplate.
- The WriteOnlyVintaScheduleModelViewSet class description says it doesn’t allow updates, yet it includes UpdateModelMixin—either update the documentation or remove the conflicting mixin to align behavior and docs.
## Individual Comments
### Comment 1
<location> `calendar_integration/services/calendar_permission_service.py:63` </location>
<code_context>
+ raise ValueError("Invalid token format")
+ token_id = token_parts[0]
+ token_str = token_parts[1]
+ except (ValueError, UnicodeDecodeError) as e:
+ raise InvalidTokenError("Invalid token format") from e
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Token decoding error handling could be more robust.
Also catch binascii.Error from base64.b64decode to prevent unhandled exceptions with invalid input.
Suggested implementation:
```python
import binascii
try:
token_full_str = base64.b64decode(token_str_base64).decode("utf-8")
token_parts = token_full_str.split(":")
if len(token_parts) != 2:
raise ValueError("Invalid token format")
token_id = token_parts[0]
token_str = token_parts[1]
```
```python
def initialize_with_token(self, token_str_base64: str, organization_id: int):
try:
token_full_str = base64.b64decode(token_str_base64).decode("utf-8")
token_parts = token_full_str.split(":")
if len(token_parts) != 2:
raise ValueError("Invalid token format")
```
```python
except (ValueError, UnicodeDecodeError, binascii.Error) as e:
raise InvalidTokenError("Invalid token format") from e
```
</issue_to_address>
### Comment 2
<location> `calendar_integration/token_views.py:92-101` </location>
<code_context>
+ super().__init__(**kwargs)
+ self.calendar_service = calendar_service
+
+ def get_serializer_context(self):
+ """
+ Provide serializer context with token information for authentication.
+ """
+ context = super().get_serializer_context()
+
+ # Only try to extract token if there's a proper Authorization header
+ auth_header = self.request.headers.get("authorization", "")
+ if auth_header.startswith("Bearer ") and len(auth_header) > 7:
+ try:
+ organization_id = self.kwargs.get("organization_id")
+ if not organization_id:
+ return context
+
+ organization = Organization.objects.get(id=organization_id)
</code_context>
<issue_to_address>
**🚨 suggestion (security):** Token context extraction may leak revoked token information.
Explicitly raise PermissionDenied when a revoked token is detected to prevent silent failures and clarify authentication errors.
</issue_to_address>
### Comment 3
<location> `calendar_integration/tests/test_token_views.py:170-179` </location>
<code_context>
+ def test_create_event_with_calendar_owner_token(
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for edge cases with missing required fields in event creation.
Please add tests for scenarios where required fields are missing or invalid to verify that the API returns correct validation errors.
</issue_to_address>
### Comment 4
<location> `calendar_integration/tests/test_token_views.py:313-322` </location>
<code_context>
+ def test_update_event_with_event_token(self, client, event_token, event, organization):
</code_context>
<issue_to_address>
**suggestion (testing):** Test for updating event with invalid/missing permissions is missing.
Add a test to verify that updates are rejected with a 403 error when the token lacks UPDATE_DETAILS or RESCHEDULE permissions.
Suggested implementation:
```python
def test_update_event_with_event_token(self, client, event_token, event, organization):
"""Test updating an event with an event-specific token."""
token, token_str = event_token
url = reverse(
"calendar_token_api:token-events-detail",
kwargs={"organization_id": organization.id, "pk": event.id},
)
update_data = {
"title": "Updated Event Title",
"description": "Updated description",
}
response = client.patch(
url,
data=json.dumps(update_data),
content_type="application/json",
HTTP_AUTHORIZATION=f"Bearer {token_str}",
)
assert response.status_code == status.HTTP_200_OK
def test_update_event_with_insufficient_permissions(self, client, event, organization, event_token_factory):
"""
Test that updating an event with a token lacking UPDATE_DETAILS or RESCHEDULE permissions is rejected.
"""
# Create a token with insufficient permissions (e.g., only VIEW)
insufficient_permissions = ["VIEW"]
token, token_str = event_token_factory(event=event, permissions=insufficient_permissions)
url = reverse(
"calendar_token_api:token-events-detail",
kwargs={"organization_id": organization.id, "pk": event.id},
)
update_data = {
"title": "Should Not Update",
"description": "Should Not Update",
}
response = client.patch(
url,
data=json.dumps(update_data),
content_type="application/json",
HTTP_AUTHORIZATION=f"Bearer {token_str}",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
```
- Ensure that `event_token_factory` is available as a fixture in your test suite. If not, you will need to implement or import it. It should allow you to create event tokens with specific permissions.
- If your permission names differ (e.g., not "VIEW", "UPDATE_DETAILS", "RESCHEDULE"), adjust them to match your codebase.
</issue_to_address>
### Comment 5
<location> `calendar_integration/tests/test_token_views.py:456-465` </location>
<code_context>
+ def test_delete_event_with_event_token(self, client, event_token, event, organization):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider testing deletion of non-existent events.
Add a test case for deleting a non-existent event to verify the API returns the correct error response, such as a 404.
</issue_to_address>
### Comment 6
<location> `calendar_integration/tests/test_token_views.py:609-480` </location>
<code_context>
+ def test_delete_event_with_series_parameter(self, client, event_token, event, organization):
</code_context>
<issue_to_address>
**suggestion (testing):** Test for deleting a series when the event is not part of a series is missing.
Add a test where delete_series is true for a non-series event to verify correct API behavior.
Suggested implementation:
```python
def test_delete_event_with_series_parameter(self, client, event_token, event, organization):
"""Test deleting an event with delete_series parameter."""
token, token_str = event_token
url = reverse(
"calendar_token_api:token-events-detail",
kwargs={"organization_id": organization.id, "pk": event.id},
)
# Mock the calendar service delete_event method
with patch(
"calendar_integration.services.calendar_service.CalendarService.delete_event"
):
response = client.delete(
url + "?delete_series=true",
HTTP_AUTHORIZATION=f"Bearer {token_str}",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
def test_delete_series_on_non_series_event(self, client, event_token, event, organization):
"""Test deleting a non-series event with delete_series=true parameter."""
token, token_str = event_token
url = reverse(
"calendar_token_api:token-events-detail",
kwargs={"organization_id": organization.id, "pk": event.id},
)
# Ensure the event is not part of a series
assert not getattr(event, "series_id", None)
with patch(
"calendar_integration.services.calendar_service.CalendarService.delete_event"
):
response = client.delete(
url + "?delete_series=true",
HTTP_AUTHORIZATION=f"Bearer {token_str}",
)
# Expecting either 400 Bad Request or 204 No Content depending on API design
# Adjust assertion as per actual API behavior
assert response.status_code in [status.HTTP_400_BAD_REQUEST, status.HTTP_204_NO_CONTENT]
```
- If your event fixture sometimes creates series events, ensure you use or create a fixture that produces a non-series event for this test.
- Adjust the expected status code in the assertion to match your API's actual behavior (e.g., 400 if it's an error, 204 if it's ignored).
</issue_to_address>
### Comment 7
<location> `calendar_integration/serializers.py:569` </location>
<code_context>
class CalendarEventSerializer(VirtualModelSerializer):
+ user: User | None
+ token: str | None
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the serializer methods by extracting context resolution, dynamic field setup, and validation logic into dedicated helper functions.
```markdown
You’ve introduced a lot of branching into `__init__`, `validate`, `create` and `update`. You can dramatically shrink each of those methods by extracting the token/user resolution, the dynamic‐fields setup, and each validation block into small, well‐named helpers. E.g.:
```python
class CalendarEventSerializer(VirtualModelSerializer):
def __init__(...):
super().__init__(...)
self._resolve_context()
self._init_dynamic_fields()
def _resolve_context(self):
ctx = self.context or {}
self.token = ctx.get("token")
self.token_str_base64 = ctx.get("token_str_base64")
self.organization = ctx.get("organization")
self.user = (
None if self.token
else getattr(ctx.get("request"), "user", None)
)
self.organization_id = (
self.organization.id
if self.organization
else getattr(self.user, "organization_membership.organization_id", None)
)
self.user_is_authenticated = bool(self.token) or (
self.user and self.user.is_authenticated
)
def _init_dynamic_fields(self):
# factor out all three identical blocks into one helper
q = lambda Model: (
Model.objects.filter_by_organization(self.organization_id).all()
if self.user_is_authenticated and self.organization_id
else Model.original_manager.none()
)
if self.instance:
self.fields["recurrence_rule_id"] = serializers.PrimaryKeyRelatedField(
source="recurrence_rule_fk",
queryset=q(RecurrenceRule),
write_only=True,
)
for name, Model in [
("google_calendar_service_account", GoogleCalendarServiceAccount),
("calendar", Calendar),
]:
self.fields[name] = serializers.PrimaryKeyRelatedField(
queryset=q(Model),
required=False,
write_only=True,
)
# nested serializers untouched...
def validate(self, attrs):
self._validate_presence_of_source(attrs)
self._validate_time_range(attrs)
self._validate_recurrence(attrs)
self._assign_default_calendar(attrs)
return attrs
def _validate_presence_of_source(self, attrs):
if (
not self.instance and
not self.token and
not any(attrs.get(k) for k in ("calendar", "provider", "google_calendar_service_account"))
):
raise serializers.ValidationError(
"You need to select either a calendar, provider, or a service account..."
)
def _validate_time_range(self, attrs):
# reuse same logic for create / update
start = attrs.get("start_time") or getattr(self.instance, "start_time_tz_unaware", None)
end = attrs.get("end_time") or getattr(self.instance, "end_time_tz_unaware", None)
if start and end and start >= end:
raise serializers.ValidationError("End time must be after start time.")
def _validate_recurrence(self, attrs):
rdata, rstring, parent = (
attrs.get("recurrence_rule"),
attrs.get("rrule_string"),
attrs.get("parent_recurring_object_id"),
)
if rdata and rstring:
raise serializers.ValidationError("Cannot specify both …")
if (rdata or rstring) and parent:
raise serializers.ValidationError("… only for master events")
def _assign_default_calendar(self, attrs):
if not self.instance and not attrs.get("calendar") and not self.token:
# same logic as before, but isolated
org_id = self.organization_id or self.user.organization_membership.organization_id
if attrs.get("provider"):
attrs["calendar"] = CalendarOwnership.objects.filter(
organization=org_id,
calendar__provider=attrs["provider"],
is_default=True,
).first()
elif attrs.get("google_calendar_service_account"):
attrs["calendar"] = attrs["google_calendar_service_account"].calendar
```
You’d apply the same pattern to `create`/`update`: pull out the “authenticate” branches into `_authenticate_token_flow` and `_authenticate_user_flow`, then each method becomes a few lines calling those helpers plus the existing pop/recurrence logic. This keeps all current functionality but makes each method easy to read and test.
</issue_to_address>
### Comment 8
<location> `calendar_integration/token_views.py:73` </location>
<code_context>
+ return calendar_service
+
+
+class TokenCalendarEventViewSet(TokenAuthenticationMixin, NoListVintaScheduleModelViewSet):
+ """
+ ViewSet for calendar event management using management tokens.
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring repeated authentication and service initialization logic into the initial() method to centralize validation and reduce duplication.
Here’s one way to collapse all of those identical try/except + `get_calendar_service(…)` calls into a single hook—override DRF’s `initial()` so it runs **before** any of your CRUD methods, and fails early if the token is invalid. You can then drop the duplicated blocks from each action:
```python
class TokenCalendarEventViewSet(TokenAuthenticationMixin, NoListVintaScheduleModelViewSet):
# … your existing attrs …
def initial(self, request, *args, **kwargs):
"""
Run token-based auth & CalendarService init once, before any action.
"""
super().initial(request, *args, **kwargs)
org_id = kwargs.get("organization_id")
if not org_id:
raise PermissionDenied("Organization ID not found in URL path")
try:
organization = Organization.objects.get(id=org_id)
except Organization.DoesNotExist:
raise PermissionDenied("Invalid organization")
# Will raise PermissionDenied if header / token is bad
self._calendar_service = self.get_calendar_service(request, organization)
# then drop all the try/except + get_calendar_service calls from:
# create, list, retrieve, update, partial_update, destroy,
# perform_create, perform_update, perform_destroy
# and in those methods just call super() or serializer.save()
def create(self, request, *args, **kwargs):
return super().create(request, *args, **kwargs)
def perform_destroy(self, instance):
delete_series = self.request.query_params.get("delete_series", "false").lower() == "true"
self._calendar_service.delete_event(
calendar_id=instance.calendar_fk_id,
event_id=instance.id,
delete_series=delete_series,
)
```
Advantages:
- All token/header/organization validation lives in one place (`initial`).
- You can store the initialized service on `self` (here `_calendar_service`) instead of re-creating it.
- CRUD methods go back to one- or two-liner overrides.
- Behavior is identical (any `PermissionDenied` raised in `initial()` returns the same 403).
</issue_to_address>
### Comment 9
<location> `calendar_integration/services/calendar_service.py:370` </location>
<code_context>
organization_id=self.organization.id
</code_context>
<issue_to_address>
**security (generic-api-key):** Detected a Generic API Key, potentially exposing access to various services and sensitive operations.
*Source: gitleaks*
</issue_to_address>
### Comment 10
<location> `calendar_integration/tests/test_token_views.py:202-204` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Avoid conditionals in tests. ([`no-conditionals-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-conditionals-in-tests))
<details><summary>Explanation</summary>Avoid complex code, like conditionals, in test functions.
Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals
Some ways to fix this:
* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.
> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.
Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>
### Comment 11
<location> `calendar_integration/tests/test_token_views.py:365-367` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Avoid conditionals in tests. ([`no-conditionals-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-conditionals-in-tests))
<details><summary>Explanation</summary>Avoid complex code, like conditionals, in test functions.
Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals
Some ways to fix this:
* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.
> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.
Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>
### Comment 12
<location> `calendar_integration/tests/test_token_views.py:602-607` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Avoid loops in tests. ([`no-loop-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-loop-in-tests))
<details><summary>Explanation</summary>Avoid complex code, like loops, in test functions.
Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals
Some ways to fix this:
* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.
> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.
Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>
### Comment 13
<location> `calendar_integration/token_views.py:80` </location>
<code_context>
authentication_classes = tuple() # Disable default authentication - we handle it manually
</code_context>
<issue_to_address>
**suggestion (code-quality):** Replace `tuple()` with `()` ([`tuple-literal`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/tuple-literal))
```suggestion
authentication_classes = () # Disable default authentication - we handle it manually
```
<br/><details><summary>Explanation</summary>The most concise and Pythonic way to create an empty tuple is to use the `()`
literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
```python
x = ("first", "second")
```
Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
```
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
```
```
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
```
</details>
</issue_to_address>
### Comment 14
<location> `calendar_integration/token_views.py:81` </location>
<code_context>
permission_classes = tuple() # Disable default permissions - we handle it manually
</code_context>
<issue_to_address>
**suggestion (code-quality):** Replace `tuple()` with `()` ([`tuple-literal`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/tuple-literal))
```suggestion
permission_classes = () # Disable default permissions - we handle it manually
```
<br/><details><summary>Explanation</summary>The most concise and Pythonic way to create an empty tuple is to use the `()`
literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
```python
x = ("first", "second")
```
Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
```
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
```
```
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
```
</details>
</issue_to_address>
### Comment 15
<location> `calendar_integration/serializers.py:795-799` </location>
<code_context>
def validate(self, attrs):
calendar = attrs.get("calendar")
# For token-based requests, we can infer the calendar from the token context
# Only check for calendar/provider/service account during creation for non-token requests
if (
not self.instance # This is a creation, not an update
and not calendar
and not attrs.get("provider")
and not attrs.get("google_calendar_service_account")
and not self.token # No token present - require explicit calendar/provider/service account
):
raise serializers.ValidationError(
"You need to select either a calendar, provider, or a service account to create "
"an event."
)
# Validate start_time and end_time only if both are provided or if this is a creation
start_time = attrs.get("start_time")
end_time = attrs.get("end_time")
# For updates, use existing instance values if not provided in attrs
if self.instance:
start_time = start_time or self.instance.start_time_tz_unaware
end_time = end_time or self.instance.end_time_tz_unaware
if start_time and end_time and start_time >= end_time:
raise serializers.ValidationError("End time must be after start time.")
# Validate recurrence fields
recurrence_rule_data = attrs.get("recurrence_rule")
rrule_string = attrs.get("rrule_string")
parent_recurring_object_id = attrs.get("parent_recurring_object_id")
if recurrence_rule_data and rrule_string:
raise serializers.ValidationError(
"Cannot specify both recurrence_rule and rrule_string. Use one or the other."
)
if (recurrence_rule_data or rrule_string) and parent_recurring_object_id:
raise serializers.ValidationError(
"Cannot specify recurrence rule for event instances. Recurrence rules are only for master events."
)
# Only auto-assign calendar during creation, not updates
if not calendar and not self.instance:
if self.token:
# For token-based requests, we'll infer the calendar in the create method
# after the calendar service is initialized, so we can skip calendar assignment here
pass
else:
# Use stored user or organization from regular authentication
if self.user and hasattr(self.user, "organization_membership"):
organization_id = self.user.organization_membership.organization_id
elif self.organization:
organization_id = self.organization.id
else:
raise serializers.ValidationError(
"Cannot determine organization for calendar selection."
)
if attrs.get("provider"):
attrs["calendar"] = CalendarOwnership.objects.filter(
organization=organization_id,
calendar__provider=attrs.get("provider"),
is_default=True,
).first()
elif attrs.get("google_calendar_service_account"):
attrs["calendar"] = attrs.get("google_calendar_service_account").calendar
return attrs
</code_context>
<issue_to_address>
**suggestion (code-quality):** Swap if/else to remove empty if body ([`remove-pass-body`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-pass-body/))
```suggestion
if not self.token:
```
</issue_to_address>
### Comment 16
<location> `calendar_integration/serializers.py:824` </location>
<code_context>
def create(self, validated_data):
if not self.calendar_service:
raise CalendarServiceNotInjectedError(
"calendar_service is not defined, please configure your DI container correctly"
)
calendar: Calendar | None = validated_data.pop("calendar", None)
# Use token or user for authentication
if self.token_str_base64:
# Token-based authentication - initialize without provider
# We need to get the organization from the token context first
organization = self.organization
if not organization:
raise serializers.ValidationError(
"Organization context is required for token-based authentication"
)
self.calendar_service.initialize_without_provider(
user_or_token=self.token_str_base64, organization=organization
)
# If no calendar was provided, infer it from the token after service initialization
if not calendar:
# The calendar service now has the permission service initialized with the token
permission_service = self.calendar_service.calendar_permission_service
if permission_service and permission_service.token:
if permission_service.token.calendar_fk:
# Calendar-level token - use the token's calendar
calendar = permission_service.token.calendar_fk
elif permission_service.token.event_fk:
# Event-level token - use the event's calendar
calendar = permission_service.token.event_fk.calendar_fk
else:
raise serializers.ValidationError(
"Unable to determine calendar from token context."
)
else:
raise serializers.ValidationError(
"Token authentication failed or calendar could not be determined."
)
elif self.user:
# Regular user authentication
if not calendar:
raise serializers.ValidationError(
"Calendar is required for user-based authentication"
)
user = self.user
if validated_data.get("google_calendar_service_account"):
account = validated_data.get("google_calendar_service_account")
self.calendar_service.authenticate(
account=account,
organization=calendar.organization,
)
else:
self.calendar_service.authenticate(
account=user,
organization=calendar.organization,
)
else:
raise serializers.ValidationError(
{
"non_field_errors": [
"You need either an authenticated user or a Calendar Management Token"
]
}
)
resource_allocations = validated_data.pop("resource_allocations", [])
attendances = validated_data.pop("attendances", [])
external_attendances = validated_data.pop("external_attendances", [])
# Handle recurrence fields
recurrence_rule_data = validated_data.pop("recurrence_rule", None)
rrule_string = validated_data.pop("rrule_string", None)
parent_recurring_object_id = validated_data.pop("parent_recurring_object_id", None)
# Prepare recurrence rule for calendar service
final_rrule_string = None
if recurrence_rule_data:
# Convert recurrence_rule_data to RRULE string
temp_rule = RecurrenceRule(organization=calendar.organization, **recurrence_rule_data)
final_rrule_string = temp_rule.to_rrule_string()
elif rrule_string:
final_rrule_string = rrule_string
event = self.calendar_service.create_event(
calendar_id=calendar.id,
event_data=CalendarEventInputData(
title=validated_data.get("title"),
description=validated_data.get("description"),
start_time=validated_data.get("start_time"),
end_time=validated_data.get("end_time"),
timezone=validated_data.get("timezone"),
resource_allocations=[
ResourceAllocationInputData(resource_id=ra["calendar"].id)
for ra in resource_allocations
],
attendances=[
EventAttendanceInputData(user_id=att["user"].id) for att in attendances
],
external_attendances=[
EventExternalAttendanceInputData(
external_attendee=ExternalAttendeeInputData(
id=ext["external_attendee"].get("id"),
email=ext["external_attendee"]["email"],
name=ext["external_attendee"]["name"],
)
)
for ext in external_attendances
],
# Recurrence fields
recurrence_rule=final_rrule_string,
parent_event_id=parent_recurring_object_id,
is_recurring_exception=validated_data.get("is_recurring_exception", False),
),
)
return event
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Inline variable that is immediately returned ([`inline-immediately-returned-variable`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/inline-immediately-returned-variable/))
- Swap if/else branches ([`swap-if-else-branches`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/swap-if-else-branches/))
- Remove unnecessary else after guard condition ([`remove-unnecessary-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-unnecessary-else/))
</issue_to_address>
### Comment 17
<location> `calendar_integration/services/calendar_permission_service.py:63` </location>
<code_context>
def initialize_with_token(self, token_str_base64: str, organization_id: int):
try:
token_full_str = base64.b64decode(token_str_base64).decode("utf-8")
token_parts = token_full_str.split(":")
if len(token_parts) != 2:
raise ValueError("Invalid token format")
token_id = token_parts[0]
token_str = token_parts[1]
except (ValueError, UnicodeDecodeError) as e:
raise InvalidTokenError("Invalid token format") from e
try:
token = (
CalendarManagementToken.objects.prefetch_related("permissions")
.select_related("calendar", "event")
.filter(organization_id=organization_id)
.get(id=token_id, revoked_at__isnull=True)
)
except CalendarManagementToken.DoesNotExist as e:
# Handle any exceptions that occur during initialization
raise InvalidTokenError("Invalid token string provided") from e
if not verify_long_lived_token(token_str, token.token_hash):
raise InvalidTokenError("Invalid token string provided") from None
self.token = token
</code_context>
<issue_to_address>
**suggestion (code-quality):** We've found these issues:
- Remove redundant exceptions from an except clause ([`remove-redundant-exception`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-exception/))
- Replace length-one exception tuple with exception ([`simplify-single-exception-tuple`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/simplify-single-exception-tuple/))
```suggestion
except ValueError as e:
```
</issue_to_address>
### Comment 18
<location> `calendar_integration/tests/test_token_views.py:333` </location>
<code_context>
def test_update_event_with_event_token(self, client, event_token, event, organization):
"""Test updating an event with an event-specific token."""
token, token_str = event_token
url = reverse(
"calendar_token_api:token-events-detail",
kwargs={"organization_id": organization.id, "pk": event.id},
)
update_data = {
"title": "Updated Event Title",
"description": "Updated description",
}
response = client.patch(
url,
data=json.dumps(update_data),
content_type="application/json",
HTTP_AUTHORIZATION=self.create_auth_header(token.id, token_str),
)
assert response.status_code == status.HTTP_200_OK
# Verify event was updated
event.refresh_from_db()
assert event.title == "Updated Event Title"
assert event.description == "Updated description"
"""Test updating an event with an event-specific token."""
token, token_str = event_token
url = reverse(
"calendar_token_api:token-events-detail",
kwargs={"organization_id": organization.id, "pk": event.id},
)
update_data = {
"title": "Updated Event Title",
"description": "Updated description",
}
# Mock the calendar service update_event method
with patch(
"calendar_integration.services.calendar_service.CalendarService.update_event"
) as mock_update_event:
mock_update_event.return_value = event
response = client.patch(
url,
data=json.dumps(update_data),
content_type="application/json",
HTTP_AUTHORIZATION=self.create_auth_header(token.id, token_str),
)
# Debug the response if it's not 200
if response.status_code != status.HTTP_200_OK:
print(f"Response status: {response.status_code}")
print(f"Response content: {response.content.decode()}")
assert response.status_code == status.HTTP_200_OK
# Verify event was updated
event.refresh_from_db()
assert event.title == "Updated Event Title"
assert event.description == "Updated description"
</code_context>
<issue_to_address>
**issue (code-quality):** Extract duplicate code into method ([`extract-duplicate-method`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/extract-duplicate-method/))
</issue_to_address>
### Comment 19
<location> `calendar_integration/token_views.py:47-49` </location>
<code_context>
def extract_token_from_header(self, request: Request) -> tuple[str, int]:
"""
Extract and validate basic token format from Authorization header.
Returns (token_str_base64, organization_id) from the URL path.
"""
auth_header = request.headers.get("authorization", "")
if not auth_header.startswith("Bearer "):
raise PermissionDenied(
"Missing or invalid Authorization header. Expected: Bearer <token>"
)
token_str_base64 = auth_header[7:] # Remove 'Bearer ' prefix
if not token_str_base64:
raise PermissionDenied("Missing token in Authorization header")
# Extract organization_id from URL path
if not request.resolver_match or not request.resolver_match.kwargs:
raise PermissionDenied("Organization ID not found in URL path")
organization_id = request.resolver_match.kwargs.get("organization_id")
if not organization_id:
raise PermissionDenied("Organization ID not found in URL path")
return token_str_base64, int(organization_id)
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Swap if/else branches ([`swap-if-else-branches`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/swap-if-else-branches/))
</issue_to_address>
### Comment 20
<location> `calendar_integration/token_views.py:129-135` </location>
<code_context>
def get_serializer_context(self):
"""
Provide serializer context with token information for authentication.
"""
context = super().get_serializer_context()
# Only try to extract token if there's a proper Authorization header
auth_header = self.request.headers.get("authorization", "")
if auth_header.startswith("Bearer ") and len(auth_header) > 7:
try:
organization_id = self.kwargs.get("organization_id")
if not organization_id:
return context
organization = Organization.objects.get(id=organization_id)
# Get the token and extract the token object
token_str_base64, _ = self.extract_token_from_header(self.request)
decoded_token = base64.b64decode(token_str_base64).decode()
token_id, _ = decoded_token.split(":", 1)
# Get the token object with organization filter
token = CalendarManagementToken.objects.filter_by_organization(organization.id).get(
id=token_id
)
# Check if token is revoked
if token.revoked_at is not None:
return context # Don't include token info for revoked tokens
# Pass token string and organization to serializer context
context["token"] = token
context[
"token_str_base64"
] = token_str_base64 # For calendar service authentication
context["organization"] = organization
except (
ValueError,
CalendarManagementToken.DoesNotExist,
Organization.DoesNotExist,
UnicodeDecodeError,
PermissionDenied,
):
# If token is invalid, silently continue without token context
# The individual HTTP methods will handle authentication and return proper error responses
pass
return context
</code_context>
<issue_to_address>
**suggestion (code-quality):** Remove redundant exceptions from an except clause ([`remove-redundant-exception`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-exception/))
```suggestion
except (ValueError, CalendarManagementToken.DoesNotExist, Organization.DoesNotExist, PermissionDenied):
```
</issue_to_address>
### Comment 21
<location> `calendar_integration/token_views.py:146-147` </location>
<code_context>
def get_queryset(self):
"""
Filter events based on organization from the URL.
"""
organization_id = self.kwargs.get("organization_id")
if organization_id:
return CalendarEvent.objects.filter_by_organization(organization_id)
return CalendarEvent.objects.none()
</code_context>
<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if organization_id := self.kwargs.get("organization_id"):
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+92
to
+101
| def get_serializer_context(self): | ||
| """ | ||
| Provide serializer context with token information for authentication. | ||
| """ | ||
| context = super().get_serializer_context() | ||
|
|
||
| # Only try to extract token if there's a proper Authorization header | ||
| auth_header = self.request.headers.get("authorization", "") | ||
| if auth_header.startswith("Bearer ") and len(auth_header) > 7: | ||
| try: |
Contributor
There was a problem hiding this comment.
🚨 suggestion (security): Token context extraction may leak revoked token information.
Explicitly raise PermissionDenied when a revoked token is detected to prevent silent failures and clarify authentication errors.
| return calendar_service | ||
|
|
||
|
|
||
| class TokenCalendarEventViewSet(TokenAuthenticationMixin, NoListVintaScheduleModelViewSet): |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider refactoring repeated authentication and service initialization logic into the initial() method to centralize validation and reduce duplication.
Here’s one way to collapse all of those identical try/except + get_calendar_service(…) calls into a single hook—override DRF’s initial() so it runs before any of your CRUD methods, and fails early if the token is invalid. You can then drop the duplicated blocks from each action:
class TokenCalendarEventViewSet(TokenAuthenticationMixin, NoListVintaScheduleModelViewSet):
# … your existing attrs …
def initial(self, request, *args, **kwargs):
"""
Run token-based auth & CalendarService init once, before any action.
"""
super().initial(request, *args, **kwargs)
org_id = kwargs.get("organization_id")
if not org_id:
raise PermissionDenied("Organization ID not found in URL path")
try:
organization = Organization.objects.get(id=org_id)
except Organization.DoesNotExist:
raise PermissionDenied("Invalid organization")
# Will raise PermissionDenied if header / token is bad
self._calendar_service = self.get_calendar_service(request, organization)
# then drop all the try/except + get_calendar_service calls from:
# create, list, retrieve, update, partial_update, destroy,
# perform_create, perform_update, perform_destroy
# and in those methods just call super() or serializer.save()
def create(self, request, *args, **kwargs):
return super().create(request, *args, **kwargs)
def perform_destroy(self, instance):
delete_series = self.request.query_params.get("delete_series", "false").lower() == "true"
self._calendar_service.delete_event(
calendar_id=instance.calendar_fk_id,
event_id=instance.id,
delete_series=delete_series,
)Advantages:
- All token/header/organization validation lives in one place (
initial). - You can store the initialized service on
self(here_calendar_service) instead of re-creating it. - CRUD methods go back to one- or two-liner overrides.
- Behavior is identical (any
PermissionDeniedraised ininitial()returns the same 403).
| and isinstance(self.user_or_token, str) | ||
| ): | ||
| self.calendar_permission_service.initialize_with_token( | ||
| self.user_or_token, organization_id=self.organization.id |
Contributor
There was a problem hiding this comment.
security (generic-api-key): Detected a Generic API Key, potentially exposing access to various services and sensitive operations.
Source: gitleaks
|
|
||
| serializer_class = CalendarEventSerializer | ||
| queryset = CalendarEvent.objects.all() | ||
| authentication_classes = tuple() # Disable default authentication - we handle it manually |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace tuple() with () (tuple-literal)
Suggested change
| authentication_classes = tuple() # Disable default authentication - we handle it manually | |
| authentication_classes = () # Disable default authentication - we handle it manually |
Explanation
The most concise and Pythonic way to create an empty tuple is to use the()literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
x = ("first", "second")Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
| serializer_class = CalendarEventSerializer | ||
| queryset = CalendarEvent.objects.all() | ||
| authentication_classes = tuple() # Disable default authentication - we handle it manually | ||
| permission_classes = tuple() # Disable default permissions - we handle it manually |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace tuple() with () (tuple-literal)
Suggested change
| permission_classes = tuple() # Disable default permissions - we handle it manually | |
| permission_classes = () # Disable default permissions - we handle it manually |
Explanation
The most concise and Pythonic way to create an empty tuple is to use the()literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
x = ("first", "second")Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
Comment on lines
+795
to
+799
| if self.token: | ||
| # For token-based requests, we'll infer the calendar in the create method | ||
| # after the calendar service is initialized, so we can skip calendar assignment here | ||
| pass | ||
| else: |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Swap if/else to remove empty if body (remove-pass-body)
Suggested change
| if self.token: | |
| # For token-based requests, we'll infer the calendar in the create method | |
| # after the calendar service is initialized, so we can skip calendar assignment here | |
| pass | |
| else: | |
| if not self.token: |
| raise ValueError("Invalid token format") | ||
| token_id = token_parts[0] | ||
| token_str = token_parts[1] | ||
| except (ValueError, UnicodeDecodeError) as e: |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): We've found these issues:
- Remove redundant exceptions from an except clause (
remove-redundant-exception) - Replace length-one exception tuple with exception (
simplify-single-exception-tuple)
Suggested change
| except (ValueError, UnicodeDecodeError) as e: | |
| except ValueError as e: |
Comment on lines
+129
to
+135
| except ( | ||
| ValueError, | ||
| CalendarManagementToken.DoesNotExist, | ||
| Organization.DoesNotExist, | ||
| UnicodeDecodeError, | ||
| PermissionDenied, | ||
| ): |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Remove redundant exceptions from an except clause (remove-redundant-exception)
Suggested change
| except ( | |
| ValueError, | |
| CalendarManagementToken.DoesNotExist, | |
| Organization.DoesNotExist, | |
| UnicodeDecodeError, | |
| PermissionDenied, | |
| ): | |
| except (ValueError, CalendarManagementToken.DoesNotExist, Organization.DoesNotExist, PermissionDenied): |
Comment on lines
+146
to
+147
| organization_id = self.kwargs.get("organization_id") | ||
| if organization_id: |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression)
Suggested change
| organization_id = self.kwargs.get("organization_id") | |
| if organization_id: | |
| if organization_id := self.kwargs.get("organization_id"): |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #13 +/- ##
==========================================
- Coverage 81.19% 81.15% -0.05%
==========================================
Files 143 145 +2
Lines 7769 7970 +201
Branches 857 884 +27
==========================================
+ Hits 6308 6468 +160
- Misses 1100 1131 +31
- Partials 361 371 +10
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Introduce token-based public scheduler endpoints for calendar event management with serializer, service, and permission layer enhancements, update routing and schema, and add integration tests
New Features:
Enhancements:
Documentation:
Tests: