kennyboy55/friend-event-rsvp
1
1
Fork 0
Code Issues 29 Pull Requests Actions Packages Projects 1 Releases 1 Activity

Improve PyLint score

Browse Source
This commit is contained in:
kennyboy55
2026-03-31 15:24:00 +02:00
parent 7a9cce3d2f
commit d0c2fe18fa
20 changed files with 881 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/extensions.json
Vendored
+4 -1
View File
@@ -1,3 +1,6 @@
{
"recommendations": ["otovo-oss.htmx-tags", "esbenp.prettier-vscode"]
"recommendations": [
"otovo-oss.htmx-tags",
"ms-python.black-formatter"
]
}
.vscode/settings.json
Vendored
+10 -10
View File
@@ -1,12 +1,12 @@
{
"editor.formatOnSave": true,
"editor.defaultFormatter": "esbenp.prettier-vscode",
"[python]": {
"editor.defaultFormatter": "esbenp.prettier-vscode",
},
"python.analysis.autoImportCompletions": true,
"python.analysis.autoFormatStrings": true,
"python.analysis.completeFunctionParens": true,
"python.analysis.enableTroubleshootMissingImports": true,
"python.analysis.autoImportCompletions": true,
"python.analysis.completeFunctionParens": true,
"python.analysis.typeCheckingMode": "basic",
"python.analysis.autoSearchPaths": true,
}
"[python]": {
"editor.defaultFormatter": "ms-python.black-formatter",
"editor.formatOnSave": true
},
}
README.md
+6
View File
@@ -38,6 +38,12 @@ Requirements (installed from Pipfile):
- Daphne
- BlackNoise
### Pylint
Make sure to have the Pipfile dev dependencies installed:
`pylint --load-plugins=pylint_django --django-settings-module=rsvpproject.settings --disable=C0111,C0301 rsvpproject events`
### Publish docker image
Build
events/admin.py
+24 -2
View File
@@ -1,9 +1,31 @@
# Register your models here.
from django.contrib import admin
from events.models import *
from events.models import (
ActivityGroup,
Comment,
Event,
Guest,
Activity,
RSVP,
ActivitySelection,
Question,
QuestionChoice,
Response,
ResponseChoice,
)
class ActivityAdmin(admin.ModelAdmin):
list_display = ('title', 'group', 'event', 'guest_limit', 'remaining_capacity', 'attendee_count', 'attendee_list',)
list_display = (
"title",
"group",
"event",
"guest_limit",
"remaining_capacity",
"attendee_count",
"attendee_list",
)
admin.site.register(Event)
admin.site.register(Activity, ActivityAdmin)
events/apps.py
+2 -2
View File
@@ -2,7 +2,7 @@ from django.apps import AppConfig
class EventsConfig(AppConfig):
name = 'events'
name = "events"
def ready(self):
import events.signals
import events.signals
events/consumers.py
+48 -49
View File
@@ -4,75 +4,74 @@ from channels.generic.websocket import AsyncWebsocketConsumer
logger = logging.getLogger(__name__)
connected_websocket_consumers = 0
CONNECTED_WEBSOCKET_CONSUMERS = 0
class ActivityConsumer(AsyncWebsocketConsumer):
async def connect(self):
global connected_websocket_consumers
global CONNECTED_WEBSOCKET_CONSUMERS
if "url_route" not in self.scope:
raise ValueError
self.event_id = self.scope["url_route"]["kwargs"]["event_id"]
logger.debug(f"Connection for Event={self.event_id}")
logger.debug("Connection for Event=%s", self.event_id)
# Subscribe to database changes for this event
await self.channel_layer.group_add(
f"activity_{self.event_id}",
self.channel_name
f"activity_{self.event_id}", self.channel_name
)
await self.accept()
connected_websocket_consumers += 1
logger.debug(f"Websocket for Event={self.event_id} opened ({connected_websocket_consumers} total)")
await self.update_connected()
async def disconnect(self, close_code):
global connected_websocket_consumers
logger.debug(f"Disconnection for Event={self.event_id}")
await self.accept()
CONNECTED_WEBSOCKET_CONSUMERS += 1
logger.debug(
"Websocket for Event=%s opened (%s total)",
self.event_id,
CONNECTED_WEBSOCKET_CONSUMERS,
)
await self.update_connected()
async def disconnect(self, code):
global CONNECTED_WEBSOCKET_CONSUMERS
logger.debug("Disconnection for Event=%s", self.event_id)
# Remove from group on disconnect
await self.channel_layer.group_discard(
f"activity_{self.event_id}",
self.channel_name
f"activity_{self.event_id}", self.channel_name
)
connected_websocket_consumers -= 1
logger.debug(f"Websocket for Event={self.event_id} closed ({connected_websocket_consumers} total)")
CONNECTED_WEBSOCKET_CONSUMERS -= 1
logger.debug(
"Websocket for Event=%s closed (%s total)",
self.event_id,
CONNECTED_WEBSOCKET_CONSUMERS,
)
await self.update_connected()
async def update_connected(self):
"""Update all connected """
"""Update all connected"""
logger.debug("Sending connected update...")
await self.channel_layer.group_send(
f"activity_{self.event_id}",
{
"type": "group.message",
"activity_json": {} # connected will be filled
}
)
await self.channel_layer.group_send(
f"activity_{self.event_id}",
{"type": "group.message", "activity_json": {}}, # connected will be filled
)
async def group_message(self, event):
global connected_websocket_consumers
logger.debug(f"Message {self.event_id}: {event}")
logger.debug("Message Event=%s: `%s`", self.event_id, event)
activity_json = event.get("activity_json")
logger.debug("Ready to send: " + str(activity_json))
logger.debug("Ready to send: %s", activity_json)
if activity_json is not None:
activity_json["connected"] = connected_websocket_consumers
activity_json["connected"] = CONNECTED_WEBSOCKET_CONSUMERS
activity_str = json.dumps(activity_json)
logger.debug("Sending: " + str(activity_str))
await self.send(text_data=activity_str)
logger.debug("Sending: %s", activity_str)
await self.send(text_data=activity_str)
events/forms.py
+7 -5
View File
@@ -3,21 +3,23 @@ from django import forms
from django.forms import Textarea
from .models import RSVP, Guest
class RSVPForm(forms.ModelForm):
class Meta:
model = RSVP
fields = ['responder_email', 'send_email_updates']
fields = ["responder_email", "send_email_updates"]
class GuestForm(forms.ModelForm):
class Meta:
model = Guest
fields = ['name', 'age_group']
fields = ["name", "age_group"]
class NotesForm(forms.ModelForm):
class Meta:
model = RSVP
fields = ['notes']
fields = ["notes"]
widgets = {
"notes": Textarea(attrs={"cols": 80, "rows": 6}),
}
events/models.py
+245 -156
View File
@@ -1,7 +1,6 @@
# models.py
import os
import uuid
from typing import Self
from datetime import timedelta
from django.db import models
from django.conf import settings
@@ -25,15 +24,20 @@ class TimestampedModel(models.Model):
def event_image_path(instance, filename):
# file will be uploaded to MEDIA_ROOT/<event_slug>/overview<.extension>
return "{0}/overview{1}".format(instance.slug, os.path.splitext(filename)[1])
extension = os.path.splitext(filename)[1]
return f"{instance.slug}/overview{extension}"
class Event(TimestampedModel):
"""The main event which contains activities.
An event can have multiple activities, invites, RSVPs, and guests.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
owner = models.ForeignKey(User, null=True, blank=True, related_name="+", on_delete=models.SET_NULL)
owner = models.ForeignKey(
User, null=True, blank=True, related_name="+", on_delete=models.SET_NULL
)
title = models.CharField(max_length=255)
slug = models.SlugField(max_length=255, db_index=True)
@@ -54,7 +58,7 @@ class Event(TimestampedModel):
cookie_expire_after_event = models.DurationField(default=timedelta(days=7))
class Meta:
ordering = ['-id']
ordering = ["-id"]
def __str__(self):
return self.title
@@ -65,21 +69,27 @@ class Event(TimestampedModel):
@property
def start_time(self):
first = self.activities.order_by('start_time').first() # pyright: ignore[reportAttributeAccessIssue]
first = self.activities.order_by(
"start_time"
).first() # pyright: ignore[reportAttributeAccessIssue]
return first.start_time if first else None
@property
def end_time(self):
last = self.activities.order_by('-end_time').first() # pyright: ignore[reportAttributeAccessIssue]
last = self.activities.order_by(
"-end_time"
).first() # pyright: ignore[reportAttributeAccessIssue]
return last.end_time if last else None
@property
def cookie_expire_datetime(self):
expiretime = self.end_time + self.cookie_expire_after_event
return expiretime
def is_locked(self) -> bool:
last = self.activities.order_by('-end_time').first() # pyright: ignore[reportAttributeAccessIssue]
last = self.activities.order_by(
"-end_time"
).first() # pyright: ignore[reportAttributeAccessIssue]
if not last:
return False
return timezone.now() >= (last.start_time - self.response_lock_before)
@@ -95,8 +105,11 @@ class Event(TimestampedModel):
class ActivityGroup(TimestampedModel):
"""Optional grouping for activities; groups can enforce single-choice selection."""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
event = models.ForeignKey(Event, related_name='activity_groups', on_delete=models.CASCADE)
event = models.ForeignKey(
Event, related_name="activity_groups", on_delete=models.CASCADE
)
title = models.CharField(max_length=255)
single_choice = models.BooleanField(default=False)
@@ -105,22 +118,33 @@ class ActivityGroup(TimestampedModel):
def activity_image_path(instance, filename):
# file will be uploaded to MEDIA_ROOT/<event_slug>/overview<.extension>
return "{0}/{1}{2}".format(instance.event.slug, instance.id, os.path.splitext(filename)[1])
# file will be uploaded to MEDIA_ROOT/<event_slug>/<id><.extension>
extension = os.path.splitext(filename)[1]
return f"{instance.event.slug}/{instance.id}{extension}"
class Activity(TimestampedModel):
"""An individual activity within an event.
Activities can be grouped into ActivityGroups. This is the main object that guests RSVP to.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
event = models.ForeignKey(Event, related_name='activities', on_delete=models.CASCADE)
group = models.ForeignKey(ActivityGroup, null=True, blank=True, related_name='activities', on_delete=models.SET_NULL)
event = models.ForeignKey(
Event, related_name="activities", on_delete=models.CASCADE
)
group = models.ForeignKey(
ActivityGroup,
null=True,
blank=True,
related_name="activities",
on_delete=models.SET_NULL,
)
title = models.CharField(max_length=255)
description = models.TextField(blank=True)
location = models.CharField(max_length=255)
start_time = models.DateTimeField()
end_time = models.DateTimeField(null=True, blank=True)
@@ -140,7 +164,7 @@ class Activity(TimestampedModel):
def is_locked(self) -> bool:
if not self.allow_rsvp:
return True
return timezone.now() >= (self.start_time - self.event.response_lock_before)
def remaining_capacity(self) -> int:
@@ -154,14 +178,14 @@ class Activity(TimestampedModel):
def attendee_list(self) -> list[list[str]]:
"""
Create a json list of names who are joining this activity
Context:
- An activity is linked to RSVPs in ActivitySelection
- An RSVP has a guests property which is a list of Guest objects for that group
- A guest can have the main_invitee property set, but not neceserilly
- Each Guest has a name property
- A name can be: "John", or "John Doe", or "John Peter Doe", or for foreign guests: "John van Doe"
Steps:
1. Get the RSVPs which are coming to Activity (using ActivitySelection)
2. For each RSVP, get the guest list
@@ -172,27 +196,25 @@ class Activity(TimestampedModel):
For example: "John D.", or "John van D."
3. Create a list of lists: [['John D.', 'Jane', 'Jack H.'], ['Maria'], ['John van D.', 'Peter'], ['Jack K.']]
4. Return this list
"""
selected_rsvps = ActivitySelection.objects.filter(activity=self).values_list('rsvp', flat=True)
selected_rsvps = ActivitySelection.objects.filter(activity=self).values_list(
"rsvp", flat=True
)
if not selected_rsvps:
return []
age_order = {
'adult': 0,
'child': 1,
'baby': 2
}
age_order = {"adult": 0, "child": 1, "baby": 2}
def sort_key(guest):
# main_invitee first (False = 0, True = 1)
main_invitee_priority = 0 if guest.is_main_invitee else 1
# Then by age group (old to young)
age_priority = age_order.get(guest.age_group, 2)
return (main_invitee_priority, age_priority)
# Store all first names
first_names_tracker = {}
for rsvp_id in selected_rsvps:
@@ -200,7 +222,7 @@ class Activity(TimestampedModel):
for guest in guests:
name_parts = guest.name.split()
first_name = name_parts[0]
if first_name in first_names_tracker:
first_names_tracker[first_name] += 1
else:
@@ -209,57 +231,62 @@ class Activity(TimestampedModel):
# Create the lists
guest_list = []
for rsvp_id in selected_rsvps:
guests_non_sort = Guest.objects.filter(rsvp=rsvp_id, )
guests_non_sort = Guest.objects.filter(
rsvp=rsvp_id,
)
guests = sorted(guests_non_sort, key=sort_key)
rsvp_list = []
for guest in guests:
name_parts = guest.name.split()
first_name = name_parts[0]
# If no duplicate and no last name, just first_name
name_to_store = first_name
# Check if name is duplicate
if first_name in first_names_tracker and first_names_tracker[first_name] > 1:
if (
first_name in first_names_tracker
and first_names_tracker[first_name] > 1
):
if len(name_parts) == 2:
name_to_store = first_name + " " + name_parts[-1][0] + "."
elif len(name_parts) > 2:
name_to_store = first_name
for name_part in name_parts[1:-1]:
name_to_store += " " + name_part
name_to_store += " " + name_parts[-1][0] + "."
rsvp_list += [name_to_store]
guest_list += [rsvp_list]
# Step 5: Return as JSON string
return guest_list
class RSVP(TimestampedModel):
class Page(models.TextChoices):
EMAIL = 'email', _('Email')
GUESTS = 'guests', _('Guests')
ACTIVITIES = 'activities', _('Activities')
QUESTIONS = 'questions', _('Questions')
NOTES = 'notes', _('Notes')
COMPLETED = 'completed', _('Completed')
EMAIL = "email", _("Email")
GUESTS = "guests", _("Guests")
ACTIVITIES = "activities", _("Activities")
QUESTIONS = "questions", _("Questions")
NOTES = "notes", _("Notes")
COMPLETED = "completed", _("Completed")
class Status(models.TextChoices):
UNOPENED = 'unopened', _('Not Opened')
NOT_STARTED = 'not_started', _('Not Started')
INCOMPLETE = 'incomplete', _('Incomplete')
NOT_COMING = 'not_coming', _('Not coming')
YES_JOINING = 'joining', _('Yes, joining')
UNOPENED = "unopened", _("Not Opened")
NOT_STARTED = "not_started", _("Not Started")
INCOMPLETE = "incomplete", _("Incomplete")
NOT_COMING = "not_coming", _("Not coming")
YES_JOINING = "joining", _("Yes, joining")
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
event = models.ForeignKey(Event, related_name='rsvps', on_delete=models.CASCADE)
event = models.ForeignKey(Event, related_name="rsvps", on_delete=models.CASCADE)
invite = models.SlugField(max_length=255, db_index=True, unique=True)
edit_token = models.UUIDField(default=uuid.uuid4, editable=False, unique=True)
@@ -267,52 +294,57 @@ class RSVP(TimestampedModel):
responder_name = models.CharField(max_length=255, blank=True)
responder_email = models.EmailField(blank=False)
send_email_updates = models.BooleanField(default=True)
personalized_message = models.TextField(blank=True)
allow_bring_guests = models.BooleanField(default=False)
page = models.CharField(max_length=16, choices=Page.choices, default=Page.EMAIL)
furthest_page = models.CharField(max_length=16, choices=Page.choices, default=Page.EMAIL)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.UNOPENED)
furthest_page = models.CharField(
max_length=16, choices=Page.choices, default=Page.EMAIL
)
status = models.CharField(
max_length=16, choices=Status.choices, default=Status.UNOPENED
)
notes = models.TextField(blank=True)
# Activities selected by this RSVP (through ActivitySelection)
activities = models.ManyToManyField(Activity, through='ActivitySelection', related_name='rsvps')
activities = models.ManyToManyField(
Activity, through="ActivitySelection", related_name="rsvps"
)
def __str__(self):
return f"RSVP {self.edit_token} — {self.event.title} — {self.responder_email} — {self.status}"
@property
def guest_count(self) -> int:
return self.guests.count() # pyright: ignore[reportAttributeAccessIssue]
return self.guests.count() # pyright: ignore[reportAttributeAccessIssue]
def selected_activities_ids(self):
selected_activities = (
ActivitySelection.objects
.filter(rsvp=self)
)
selected_activities = ActivitySelection.objects.filter(rsvp=self)
selected_activities_ids = [act.activity.id for act in selected_activities]
return selected_activities_ids
@property
def name(self) -> str:
main_guest = self.guests.filter(is_main_invitee=True).first() # pyright: ignore[reportAttributeAccessIssue]
if(main_guest):
main_guest = self.guests.filter(
is_main_invitee=True
).first() # pyright: ignore[reportAttributeAccessIssue]
if main_guest:
return main_guest.name
if(self.responder_name):
if self.responder_name:
return self.responder_name
return "friend"
PAGE_ORDER = [
'email',
'guests',
'activities',
'questions',
'notes',
'completed',
"email",
"guests",
"activities",
"questions",
"notes",
"completed",
]
def next_page(self) -> Page:
@@ -327,28 +359,28 @@ class RSVP(TimestampedModel):
except ValueError:
# If page is not in the list (e.g., INVALID), start from EMAIL
pass
if is_next_page(new_page, self.Page(self.furthest_page)):
self.furthest_page = new_page
if self.status == self.Status.NOT_STARTED:
#TODO send an email here
# TODO send an email here
self.status = self.Status.INCOMPLETE
self.page = new_page
if self.page == self.Page.COMPLETED:
self.status = self.Status.YES_JOINING
self.save()
return new_page
def previous_page(self) -> Page:
"""Move to the previous page and save to model"""
new_page = self.page
try:
current_index = self.PAGE_ORDER.index(self.page)
if current_index > 0:
@@ -356,72 +388,93 @@ class RSVP(TimestampedModel):
except ValueError:
# If page is not in the list (e.g., INVALID), stay at INVALID
pass
self.page = new_page
self.save()
return new_page
def set_page(self, page:str) -> Page:
def set_page(self, page: str) -> Page:
"""Move to the specific page and save to model"""
new_page = self.page
try:
new_page = self.Page(page)
except ValueError:
pass
self.page = new_page
self.save()
return new_page
def is_page_finished(self, page:Page) -> bool:
def is_page_finished(self, page: Page) -> bool:
"""return `True` if page is previous to the furthest reached page"""
return is_previous_page(page, self.Page(self.furthest_page))
def is_next_page_allowed(self) -> bool | str:
def is_next_page_allowed(self) -> tuple[bool, str]:
# Activities page: require at least one selected activity
if self.page == self.Page.ACTIVITIES:
if not self.activities.exists():
return "You must choose at least one activity."
return True
return False, "You must choose at least one activity."
return True, ""
# Questions page: ensure every required question for selected activities is answered
if self.page == self.Page.QUESTIONS:
selected_activities = self.activities.all()
required_questions = Question.objects.filter(activity__in=selected_activities, required=True)
required_questions = Question.objects.filter(
activity__in=selected_activities, required=True
)
for q in required_questions:
if q.scope == "rsvp":
resp_qs = Response.objects.filter(rsvp=self, question=q)
if not resp_qs.exists():
return f'Please answer required question: "{q.text}"'
return False, f'Please answer required question: "{q.text}"'
for resp in resp_qs:
if q.kind in ("text", "integer", "boolean"):
if not (resp.value_text and resp.value_text.strip()):
return f'Please answer required question: "{q.text}"'
return (
False,
f'Please answer required question: "{q.text}"',
)
else: # single / multiple choice
if resp.choices.count() == 0:
return f'Please answer required question: "{q.text}"'
return (
False,
f'Please answer required question: "{q.text}"',
)
elif q.scope == "guest":
for guest in self.guests.all(): # pyright: ignore[reportAttributeAccessIssue]
for (
guest
) in (
self.guests.all()
): # pyright: ignore[reportAttributeAccessIssue]
resp = Response.objects.filter(guest=guest, question=q).first()
if not resp:
return f'Please answer required question "{q.text}" for all guests.'
return (
False,
f'Please answer required question "{q.text}" for all guests.',
)
if q.kind in ("text", "integer", "boolean"):
if not (resp.value_text and resp.value_text.strip()):
return f'Please answer required question "{q.text}" for all guests.'
return (
False,
f'Please answer required question "{q.text}" for all guests.',
)
else:
if resp.choices.count() == 0:
return f'Please answer required question "{q.text}" for all guests.'
return (
False,
f'Please answer required question "{q.text}" for all guests.',
)
return True
return True, ""
# Default: allow moving forward
return True
return True, ""
class ActivitySelection(models.Model):
@@ -431,87 +484,102 @@ class ActivitySelection(models.Model):
because cross-table unique constraints involving a related table's column can't be expressed simply
as a DB constraint here.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
rsvp = models.ForeignKey(RSVP, related_name='activity_selections', on_delete=models.CASCADE)
activity = models.ForeignKey(Activity, related_name='activity_selections', on_delete=models.CASCADE)
rsvp = models.ForeignKey(
RSVP, related_name="activity_selections", on_delete=models.CASCADE
)
activity = models.ForeignKey(
Activity, related_name="activity_selections", on_delete=models.CASCADE
)
class Meta:
unique_together = (('rsvp', 'activity'),)
unique_together = (("rsvp", "activity"),)
def clean(self):
# If the activity belongs to a group that is single_choice, ensure the RSVP has no other selections
group = self.activity.group
if group and group.single_choice:
other = ActivitySelection.objects.filter(rsvp=self.rsvp, activity__group=group).exclude(pk=self.pk)
other = ActivitySelection.objects.filter(
rsvp=self.rsvp, activity__group=group
).exclude(pk=self.pk)
if other.exists():
raise ValidationError(f"You may only select one activity from the group '{group.title}'.")
raise ValidationError(
f"You may only select one activity from the group '{group.title}'."
)
class Guest(TimestampedModel):
AGE_GROUP = [
("baby", "Baby (0-2)"),
("child", "Child (3-15)"),
("adult", "Adult (15+)")
("adult", "Adult (15+)"),
]
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
event = models.ForeignKey(Event, related_name='guests', on_delete=models.CASCADE)
rsvp = models.ForeignKey(RSVP, related_name='guests', on_delete=models.CASCADE)
event = models.ForeignKey(Event, related_name="guests", on_delete=models.CASCADE)
rsvp = models.ForeignKey(RSVP, related_name="guests", on_delete=models.CASCADE)
name = models.CharField(max_length=255)
age_group = models.CharField(max_length=10, choices=AGE_GROUP, default='adult')
age_group = models.CharField(max_length=10, choices=AGE_GROUP, default="adult")
is_main_invitee = models.BooleanField(default=False)
created_by_host = models.BooleanField(default=False)
def __str__(self):
return f"{self.name} ({self.age_group})"
@property
def firstname(self) -> str:
return self.name.split(" ")[0]
return self.name.split(" ", maxsplit=1)[0]
@property
def lastname(self) -> str:
return self.name.split(" ")[1:]
return self.name.rsplit(" ", maxsplit=1)[-1]
class Question(TimestampedModel):
KIND_CHOICES = [
('text', 'Text'),
('single', 'Single choice'),
('multiple', 'Multiple choice'),
('integer', 'Integer'),
('boolean', 'Yes/No')
("text", "Text"),
("single", "Single choice"),
("multiple", "Multiple choice"),
("integer", "Integer"),
("boolean", "Yes/No"),
]
SCOPE_CHOICES = [
("rsvp", "Once per RSVP"),
("guest", "Once per Guest"),
]
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
activity = models.ForeignKey(Activity, related_name='questions', on_delete=models.CASCADE)
activity = models.ForeignKey(
Activity, related_name="questions", on_delete=models.CASCADE
)
text = models.CharField(max_length=1000)
help_text = models.CharField(max_length=1000, blank=True)
kind = models.CharField(max_length=10, choices=KIND_CHOICES, default='text')
kind = models.CharField(max_length=10, choices=KIND_CHOICES, default="text")
scope = models.CharField(max_length=10, choices=SCOPE_CHOICES, default="rsvp")
required = models.BooleanField(default=False)
order = models.PositiveIntegerField(default=0)
def __str__(self):
return f"Q: {self.text} (on `{self.activity} - {self.scope}` )"
class QuestionChoice(models.Model):
question = models.ForeignKey(Question, related_name='choices', on_delete=models.CASCADE)
question = models.ForeignKey(
Question, related_name="choices", on_delete=models.CASCADE
)
value = models.CharField(max_length=255)
label = models.CharField(max_length=255)
def __str__(self):
return self.label
class Response(TimestampedModel):
"""Answer to a Question. Questions live on Activities.
@@ -521,15 +589,26 @@ class Response(TimestampedModel):
To keep the model simple, question-level scoping is enforced at form/validation time.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
activity = models.ForeignKey(Activity, related_name='responses', on_delete=models.CASCADE)
question = models.ForeignKey(Question, related_name='responses', on_delete=models.CASCADE)
rsvp = models.ForeignKey(RSVP, null=True, blank=True, related_name='responses', on_delete=models.CASCADE)
guest = models.ForeignKey(Guest, null=True, blank=True, related_name='responses', on_delete=models.CASCADE)
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
activity = models.ForeignKey(
Activity, related_name="responses", on_delete=models.CASCADE
)
question = models.ForeignKey(
Question, related_name="responses", on_delete=models.CASCADE
)
rsvp = models.ForeignKey(
RSVP, null=True, blank=True, related_name="responses", on_delete=models.CASCADE
)
guest = models.ForeignKey(
Guest, null=True, blank=True, related_name="responses", on_delete=models.CASCADE
)
# Choice (single) if applicable
choices = models.ManyToManyField(QuestionChoice, through="ResponseChoice", related_name="responses", blank=True)
choices = models.ManyToManyField(
QuestionChoice, through="ResponseChoice", related_name="responses", blank=True
)
value_text = models.TextField(blank=True)
def clean(self):
@@ -550,7 +629,9 @@ class Response(TimestampedModel):
pass
else:
if self.choices.count() != 1:
raise ValidationError("Single-choice question requires exactly one choice.")
raise ValidationError(
"Single-choice question requires exactly one choice."
)
if kind == "multiple":
if not self.pk:
pass
@@ -563,22 +644,29 @@ class Response(TimestampedModel):
if self.question.activity.id != self.activity.id:
raise ValidationError("Question must belong to the same activity.")
def __str__(self):
def __str__(self) -> str:
whom = f"{self.question.scope}:"
if(self.question.scope == "rsvp"):
if self.question.scope == "rsvp":
whom += " " + self.rsvp.responder_name
elif(self.question.scope == "guest"):
elif self.question.scope == "guest":
whom += " " + self.guest.name
if self.question.kind in ("text", "integer", "boolean"):
return f"Response to {self.question} by {whom}: {self.value_text}"
if self.question.kind in ("single", "multiple") and self.choices.count() > 0:
return f"Response to {self.question} by {whom}: {self.choices.all}"
return f"Response to {self.question}"
class ResponseChoice(TimestampedModel):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
response = models.ForeignKey("Response", related_name="response_choices", on_delete=models.CASCADE)
choice = models.ForeignKey(QuestionChoice, related_name="response_choices", on_delete=models.CASCADE)
response = models.ForeignKey(
"Response", related_name="response_choices", on_delete=models.CASCADE
)
choice = models.ForeignKey(
QuestionChoice, related_name="response_choices", on_delete=models.CASCADE
)
class Meta:
unique_together = (("response", "choice"),)
@@ -586,12 +674,13 @@ class ResponseChoice(TimestampedModel):
def __str__(self):
return f"{self.choice} for {self.response}"
class Comment(TimestampedModel):
event = models.ForeignKey(Event, related_name='comments', on_delete=models.CASCADE)
guest = models.ForeignKey(Guest, related_name='comments', on_delete=models.CASCADE)
event = models.ForeignKey(Event, related_name="comments", on_delete=models.CASCADE)
guest = models.ForeignKey(Guest, related_name="comments", on_delete=models.CASCADE)
text = models.TextField()
notify_subscribers = models.BooleanField(default=False)
def __str__(self):
return f"Comment by {self.guest.name} on {self.event.title}"
return f"Comment by {self.guest.name} on {self.event.title}"
events/routing.py
+5 -2
View File
@@ -4,5 +4,8 @@ from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r"ws/event/(?P<event_id>[0-9a-f-]{8}-[0-9a-f-]{4}-[0-9a-f-]{4}-[0-9a-f-]{4}-[0-9a-f-]{12})/$", consumers.ActivityConsumer.as_asgi()),
]
re_path(
r"ws/event/(?P<event_id>[0-9a-f-]{8}-[0-9a-f-]{4}-[0-9a-f-]{4}-[0-9a-f-]{4}-[0-9a-f-]{12})/$",
consumers.ActivityConsumer.as_asgi(),
),
]
events/signals.py
+57 -53
View File
@@ -2,7 +2,7 @@ import logging
from asgiref.sync import async_to_sync
from django.db.models.signals import post_save, post_delete, pre_delete
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from channels.layers import get_channel_layer
from events.models import RSVP, Activity, Event, ActivitySelection, Guest, Response
@@ -14,141 +14,145 @@ logger = logging.getLogger(__name__)
# Guests
###
@receiver(post_delete, sender=Guest)
def guest_deleted(sender, instance, **kwargs):
"""Delete question responses for this guest"""
guest:Guest = instance
logger.debug(f"Signal {guest}")
guest: Guest = instance
logger.debug("Signal guest=`%s`", guest)
guest_delete_responses(guest)
guest_notify_activities(guest)
def guest_delete_responses(guest:Guest):
def guest_delete_responses(guest: Guest):
# Remove all responses from the guest that is deleted
Response.objects.filter(guest=guest).delete()
def guest_notify_activities(guest:Guest):
def guest_notify_activities(guest: Guest):
"""Broadcast activity selection changes to all connected clients."""
rsvp:RSVP = guest.rsvp
event:Event = rsvp.event
rsvp: RSVP = guest.rsvp
event: Event = rsvp.event
# Get all activities with this RSVP
activities = rsvp.activities
for activity in activities.all():
logger.debug(f"Signal {activity}")
logger.debug("Signal activity=`%s`", activity)
capacity = activity.remaining_capacity()
if capacity is None:
capacity = 100
activity_json = {
"activity": str(activity.id),
"rsvp": str(rsvp.id),
"action": "select",
"attendees": activity.attendee_count(),
"capacity": capacity,
"guests": activity.attendee_list()
"guests": activity.attendee_list(),
}
# Send message to the group
async_to_sync(send_channel)(event.id, activity_json)
@receiver(post_save, sender=Guest)
def activity_guests_changed(sender, instance, created, **kwargs):
"""Broadcast activity selection changes to all connected clients."""
guest:Guest = instance
rsvp:RSVP = guest.rsvp
event:Event = rsvp.event
logger.debug(f"Guest added: {guest.name}")
guest: Guest = instance
rsvp: RSVP = guest.rsvp
event: Event = rsvp.event
logger.debug("Guest added: name=`%s`", guest.name)
# Get all activities with this RSVP
activities = rsvp.activities
for activity in activities.all():
logger.debug(f"Signal {activity}")
logger.debug("Signal activity=`%s`", activity)
capacity = activity.remaining_capacity()
if capacity is None:
capacity = 100
activity_json = {
"activity": str(activity.id),
"rsvp": str(rsvp.id),
"action": "select",
"attendees": activity.attendee_count(),
"capacity": capacity,
"guests": activity.attendee_list()
"guests": activity.attendee_list(),
}
# Send message to the group
async_to_sync(send_channel)(event.id, activity_json)
###
# Activity Selections
###
@receiver(post_save, sender=ActivitySelection)
def activity_selection_changed(sender, instance, created, **kwargs):
"""Broadcast activity selection changes to all connected clients."""
activity:Activity = instance.activity
rsvp:RSVP = instance.rsvp
event:Event = activity.event
logger.debug(f"Signal {activity}")
activity: Activity = instance.activity
rsvp: RSVP = instance.rsvp
event: Event = activity.event
logger.debug("Signal activity=`%s`", activity)
capacity = activity.remaining_capacity()
if capacity is None:
capacity = 100
activity_json = {
"activity": str(activity.id),
"rsvp": str(rsvp.id),
"action": "select",
"attendees": activity.attendee_count(),
"capacity": capacity,
"guests": activity.attendee_list()
"guests": activity.attendee_list(),
}
# Send message to the group
async_to_sync(send_channel)(event.id, activity_json)
@receiver(post_delete, sender=ActivitySelection)
def activity_selection_deleted(sender, instance, **kwargs):
"""Broadcast activity selection changes to all connected clients."""
activity:Activity = instance.activity
rsvp:RSVP = instance.rsvp
event:Event = activity.event
logger.debug(f"Signal {activity}")
activity: Activity = instance.activity
rsvp: RSVP = instance.rsvp
event: Event = activity.event
logger.debug("Signal activity=`%s`", activity)
capacity = activity.remaining_capacity()
if capacity is None:
capacity = 100
activity_json = {
"activity": str(activity.id),
"rsvp": str(rsvp.id),
"action": "remove",
"attendees": activity.attendee_count(),
"capacity": capacity,
"guests": activity.attendee_list()
"guests": activity.attendee_list(),
}
# Send message to the group
async_to_sync(send_channel)(event.id, activity_json)
async def send_channel(event_id, activity_json):
channel_layer = get_channel_layer()
if channel_layer:
await channel_layer.group_send(
f"activity_{event_id}",
{
"type": "group.message",
"activity_json": activity_json
}
)
f"activity_{event_id}",
{"type": "group.message", "activity_json": activity_json},
)
events/urls.py
+16 -35
View File
@@ -1,95 +1,76 @@
# urls.py
from django.urls import path
from events.views.rsvp import post_step_activities
from . import views
app_name = "events"
urlpatterns = [
# Main event page
path("events/<slug:slug>/", views.event, name="event"),
path("events/<slug:slug>/", views.event_page, name="event"),
# Invite URL, can only be used once
path("invite/<slug:slug>", views.invite_redirect, name="invite"),
# RSVP Form
path("rsvp/<uuid:code>/page/<str:page>", views.rsvp_base_page, name="rsvp-base"),
path("htmx/rsvp/<str:id>/page/<str:page>", views.rsvp_pages, name="rsvp-pages"),
path("htmx/rsvp/<str:rid>/page/<str:page>", views.rsvp_pages, name="rsvp-pages"),
path(
"htmx/rsvp/<str:id>/navigate/<str:page>",
"htmx/rsvp/<str:rid>/navigate/<str:page>",
views.page_change,
name="rsvp-page-change",
),
path(
"htmx/rsvp/<str:id>/post/email",
"htmx/rsvp/<str:rid>/post/email",
views.post_step_email,
name="rsvp-post-email",
),
path(
"htmx/rsvp/<str:id>/post/guest",
views.post_step_guest,
name="rsvp-post-guest"
"htmx/rsvp/<str:rid>/post/guest", views.post_step_guest, name="rsvp-post-guest"
),
path(
"htmx/rsvp/<str:id>/delete/guest/<str:gid>",
"htmx/rsvp/<str:rid>/delete/guest/<str:gid>",
views.delete_step_guest,
name="rsvp-delete-guest"
name="rsvp-delete-guest",
),
path(
"htmx/rsvp/<str:id>/post/activity/<str:aid>",
"htmx/rsvp/<str:rid>/post/activity/<str:aid>",
views.post_step_activities,
name="rsvp-post-activity"
name="rsvp-post-activity",
),
# POST response for RSVP-scoped question
path(
"htmx/rsvp/<str:id>/post/question/<str:qid>/",
"htmx/rsvp/<str:rid>/post/question/<str:qid>/",
views.post_response_rsvp,
name="rsvp-post-response",
),
# POST response for Guest-scoped question
path(
"htmx/rsvp/<str:id>/post/question/<str:qid>/guest/<str:gid>/",
"htmx/rsvp/<str:rid>/post/question/<str:qid>/guest/<str:gid>/",
views.post_response_guest,
name="rsvp-post-response-guest",
),
path(
"htmx/rsvp/<str:id>/post/notes",
"htmx/rsvp/<str:rid>/post/notes",
views.post_step_notes,
name="rsvp-post-notes",
),
path(
"htmx/rsvp/<str:id>/start",
"htmx/rsvp/<str:rid>/start",
views.rsvp_start,
name="rsvp-start",
),
path(
"htmx/rsvp/<str:id>/restart",
"htmx/rsvp/<str:rid>/restart",
views.rsvp_restart,
name="rsvp-restart",
),
path(
"htmx/rsvp/<str:id>/update",
"htmx/rsvp/<str:rid>/update",
views.rsvp_update,
name="rsvp-update",
),
path(
"htmx/rsvp/<str:id>/decline",
"htmx/rsvp/<str:rid>/decline",
views.rsvp_decline,
name="rsvp-decline",
),
]
events/utils/cookieutils.py
+7 -7
View File
@@ -11,14 +11,14 @@ def set_rsvp_cookie(response:HttpResponse, event:Event, rsvp:RSVP):
response.set_cookie(key=event.slug, value=rsvp.edit_token, expires=event.cookie_expire_datetime)
def get_rsvp_from_cookie(request:HttpRequest, slug:str) -> RSVP | bool:
logger.debug(f"Cookies: {request.COOKIES}")
logger.debug("Cookies: %s", request.COOKIES)
edit_token = request.COOKIES.get(slug)
logger.debug(f"RSVP Edit Token: {edit_token}")
logger.debug("RSVP Edit Token: %s", edit_token)
rsvp = RSVP.objects.filter(edit_token=edit_token).first()
if rsvp is not None:
return rsvp
return False
return False
events/utils/pageutils.py
+12 -11
View File
@@ -1,16 +1,16 @@
PAGE_ORDER = [
'invalid',
'email',
'guests',
'activities',
'questions',
'notes',
'completed',
"invalid",
"email",
"guests",
"activities",
"questions",
"notes",
"completed",
]
### PAGES
def is_previous_page(first:str, second:str) -> bool:
def is_previous_page(first: str, second: str) -> bool:
"""return True if `first` is a previous page of `second`"""
try:
first_idx = PAGE_ORDER.index(first)
@@ -20,7 +20,8 @@ def is_previous_page(first:str, second:str) -> bool:
# If either page is not in the list, return False
return False
def is_next_page(first:str, second:str) -> bool:
def is_next_page(first: str, second: str) -> bool:
"""return True if `first` is a next page of `second`"""
try:
first_idx = PAGE_ORDER.index(first)
@@ -28,4 +29,4 @@ def is_next_page(first:str, second:str) -> bool:
return first_idx > second_idx
except ValueError:
# If either page is not in the list, return False
return False
return False
events/utils/rsvputils.py
+37 -36
View File
@@ -1,9 +1,7 @@
import json
from events.models import RSVP, Activity, ActivitySelection, Event
from django.core.exceptions import ValidationError
from ..models import RSVP, Activity, ActivitySelection, Event
def create_or_switch_activity_response(rsvp: RSVP, activity: Activity):
def create_or_switch_activity_response(rsvp: RSVP, activity: Activity) -> bool:
"""
This function will:
1. Check if there already is a ActivitySelection for this pair
@@ -14,78 +12,81 @@ def create_or_switch_activity_response(rsvp: RSVP, activity: Activity):
- then remove the old one
- add the new selection (switch choice)
"""
# Step 1: Remove existing selection for this activity pair (unselecting)
existing_selection = ActivitySelection.objects.filter(rsvp=rsvp, activity=activity).first()
existing_selection = ActivitySelection.objects.filter(
rsvp=rsvp, activity=activity
).first()
if existing_selection:
existing_selection.delete()
return
return False
# Step 2: Check if adding guests will fit in the selected activity
remaining_capacity = activity.remaining_capacity()
if(remaining_capacity is None):
if remaining_capacity is None:
# Hardcoded, because why not
remaining_capacity = 100
if rsvp.guest_count > remaining_capacity:
return
return False
# Step 3: Check if new selection is illegal (choosing one in same group)
# Get the activity's group if it exists
activity_group = activity.group
if activity_group and activity_group.single_choice:
# Get all other activities in the same group that are already selected
other_selections = ActivitySelection.objects.filter(
rsvp=rsvp,
activity__group=activity_group
rsvp=rsvp, activity__group=activity_group
).exclude(activity=activity)
if other_selections.exists():
# Remove the old selection (switch choice)
for other_selection in other_selections:
other_selection.delete()
# Create and save the new selection
selection = ActivitySelection(rsvp=rsvp, activity=activity)
selection.save()
return selection
def create_context(rsvp:RSVP, error:str="") -> dict:
context = {
"rsvp": rsvp,
"event": rsvp.event
}
if(error != ""):
return True
def create_context(rsvp: RSVP, error: str = "") -> dict:
context = {"rsvp": rsvp, "event": rsvp.event}
if error != "":
context["page_error"] = error
return context
def add_to_context(context:dict, key:str, value) -> dict:
def add_to_context(context: dict, key: str, value) -> dict:
context[key] = value
return context
def get_rsvp(context:dict) -> RSVP:
def get_rsvp(context: dict) -> RSVP:
rsvp = get_from_context(context, "rsvp")
if not rsvp:
raise KeyError
return rsvp
def get_event(context:dict) -> Event:
def get_event(context: dict) -> Event:
event = get_from_context(context, "event")
if not event:
raise KeyError
return event
def get_from_context(context:dict, key:str):
def get_from_context(context: dict, key: str):
if key in context:
return context[key]
return False
return False
events/utils/summary.py
+36 -29
View File
@@ -1,50 +1,55 @@
from events.models import RSVP, Activity, Question, Response
def summary_email(rsvp:RSVP) -> str:
def summary_email(rsvp: RSVP) -> str:
summary = f"Email: You will receive updates at `{rsvp.responder_email}`"
if rsvp.send_email_updates == False:
summary = f"Email: You do not want any updates."
if rsvp.send_email_updates is False:
summary = "Email: You do not want any updates."
return summary
def summary_guests(rsvp:RSVP) -> str:
summary = f"Guests: let us know who will come!"
def summary_guests(rsvp: RSVP) -> str:
summary = "Guests: let us know who will come!"
if rsvp.is_page_finished(RSVP.Page.GUESTS):
guests = rsvp.guest_count
if guests == 0:
summary = f"Guests: Don't forget to add yourself!"
summary = "Guests: Don't forget to add yourself!"
if guests == 1:
summary = f"Guests: Just you."
summary = "Guests: Just you."
if guests > 1:
summary = f"Guests: You are coming with {guests} people."
return summary
def summary_activities(rsvp:RSVP) -> str:
summary = f"Activities"
def summary_activities(rsvp: RSVP) -> str:
summary = "Activities"
chosen_activities_count = rsvp.activities.count()
available_activities = Activity.objects.filter(event=rsvp.event)
total_activities_count = len(available_activities)
activities_with_space_count = sum(
1 for a in available_activities if a.remaining_capacity() >= rsvp.guest_count
)
if rsvp.is_page_finished(RSVP.Page.ACTIVITIES) and chosen_activities_count > 0:
if chosen_activities_count == 1:
summary = f"Activities: You have selected one activity!"
summary = "Activities: You have selected one activity!"
if chosen_activities_count > 1:
summary = f"Activities: You have selected {chosen_activities_count} activities!"
summary = (
f"Activities: You have selected {chosen_activities_count} activities!"
)
else:
summary = f"Activities: You can still choose from {activities_with_space_count} / {total_activities_count} activities!"
return summary
def summary_questions(rsvp:RSVP) -> str:
def summary_questions(rsvp: RSVP) -> str:
selected_activities = rsvp.activities.all()
if not selected_activities.exists():
return "Questions: No activities selected — no questions to answer."
@@ -80,8 +85,9 @@ def summary_questions(rsvp:RSVP) -> str:
answered = sum(1 for q in questions if is_answered(q))
unanswered = total - answered
required_total = questions.filter(required=True).count()
required_unanswered = sum(1 for q in questions.filter(required=True) if not is_answered(q))
required_unanswered = sum(
1 for q in questions.filter(required=True) if not is_answered(q)
)
if rsvp.is_page_finished(RSVP.Page.QUESTIONS):
if unanswered == 0:
@@ -96,10 +102,11 @@ def summary_questions(rsvp:RSVP) -> str:
return summary
def summary_notes(rsvp:RSVP) -> str:
summary = f"Notes: anything you want to share?"
if len(rsvp.notes) > 1:
summary = f"Notes: you have left a note."
return summary
def summary_notes(rsvp: RSVP) -> str:
summary = "Notes: anything you want to share?"
if len(rsvp.notes) > 1:
summary = "Notes: you have left a note."
return summary
events/views/__init__.py
+1 -1
View File
@@ -1,2 +1,2 @@
from .views import *
from .rsvp import *
from .rsvp import *
events/views/rsvp.py
+246 -177
View File
@@ -1,27 +1,35 @@
# views.py
import logging
import re
from django.forms import ValidationError
from django.shortcuts import get_object_or_404, render, redirect
from django.db import transaction
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseForbidden
from django.http import HttpResponse, HttpResponseBadRequest
from django.template.loader import get_template
from django.urls import reverse
from django.contrib import messages
from events.utils.cookieutils import set_rsvp_cookie
from events.utils.summary import summary_activities, summary_email, summary_guests, summary_notes, summary_questions
from events.utils.summary import (
summary_activities,
summary_email,
summary_guests,
summary_notes,
summary_questions,
)
from ..utils.rsvputils import add_to_context, create_or_switch_activity_response, create_context, get_event, get_from_context, get_rsvp
from events.utils.rsvputils import (
add_to_context,
create_or_switch_activity_response,
create_context,
get_event,
get_rsvp,
)
from ..forms import GuestForm, NotesForm, RSVPForm
from events.forms import GuestForm, NotesForm, RSVPForm
from ..models import (
Event,
from events.models import (
Guest,
Activity,
ActivityGroup,
RSVP,
ActivitySelection,
Question,
@@ -33,19 +41,21 @@ from ..models import (
# Logger
logger = logging.getLogger(__name__)
# Start a new or unfinished RSVP
def rsvp_start(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
if not (rsvp.status == RSVP.Status.NOT_STARTED or rsvp.status == RSVP.Status.INCOMPLETE):
# Start a new or unfinished RSVP
def rsvp_start(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
if rsvp.status not in (RSVP.Status.NOT_STARTED, RSVP.Status.INCOMPLETE):
return HttpResponseBadRequest()
response = redirect(reverse('events:rsvp-base', args=[rsvp.edit_token, rsvp.page]))
response = redirect(reverse("events:rsvp-base", args=[rsvp.edit_token, rsvp.page]))
return response
# Update a completed RSVP
def rsvp_update(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
def rsvp_update(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
if rsvp.status != RSVP.Status.YES_JOINING:
return HttpResponseBadRequest()
@@ -53,41 +63,42 @@ def rsvp_update(request, id:str):
rsvp.page = RSVP.Page.ACTIVITIES
rsvp.save()
response = redirect(reverse('events:rsvp-base', args=[rsvp.edit_token, rsvp.page]))
response = redirect(reverse("events:rsvp-base", args=[rsvp.edit_token, rsvp.page]))
return response
# Stop an existing RSVP
def rsvp_decline(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
def rsvp_decline(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
rsvp.status = RSVP.Status.NOT_COMING
rsvp.page = RSVP.Page.EMAIL
rsvp.furthest_page = RSVP.Page.EMAIL
rsvp.save()
# Remove any current activity selection
ActivitySelection.objects.filter(rsvp=rsvp).delete()
# Remove any question responses
Response.objects.filter(rsvp=rsvp).delete()
Response.objects.filter(guest__rsvp=rsvp).delete()
response = redirect(reverse('events:event', args=[rsvp.event.slug]))
response = redirect(reverse("events:event", args=[rsvp.event.slug]))
return response
def rsvp_restart(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
def rsvp_restart(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
rsvp.status = RSVP.Status.INCOMPLETE
rsvp.save()
response = redirect(reverse('events:rsvp-base', args=[rsvp.edit_token, rsvp.page]))
response = redirect(reverse("events:rsvp-base", args=[rsvp.edit_token, rsvp.page]))
return response
# Main entrypoint for rsvp page
def rsvp_base_page(request, code:str, page:str):
def rsvp_base_page(request, code: str, page: str):
rsvp = get_object_or_404(RSVP, edit_token=code)
context = {
@@ -95,7 +106,7 @@ def rsvp_base_page(request, code:str, page:str):
"event": rsvp.event,
"invite": rsvp.invite,
"page": page,
"is_locked": rsvp.event.is_locked()
"is_locked": rsvp.event.is_locked(),
}
response = render(request, "rsvp/rsvp.html", context)
@@ -104,10 +115,10 @@ def rsvp_base_page(request, code:str, page:str):
# Render selector based on the current step of the RSVP process
def rsvp_pages(request, id:str, page:str, error:str=""):
logger.debug(f"rsvp_pages: id={id}, page={page}")
rsvp = get_object_or_404(RSVP, id=id)
def rsvp_pages(request, rid: str, page: str, error: str = ""):
logger.debug("rsvp_pages: id=%s, page=%s", rid, page)
rsvp = get_object_or_404(RSVP, id=rid)
if page == "":
page = rsvp.page
@@ -119,105 +130,111 @@ def rsvp_pages(request, id:str, page:str, error:str=""):
"page_guest": _render_page_guest(request, context, page),
"page_activities": _render_page_activities(request, context, page),
"page_questions": _render_page_questions(request, context, page),
"page_notes": _render_page_notes(request, context, page)
"page_notes": _render_page_notes(request, context, page),
}
response = render(request, "rsvp/rsvp_pages.html", rendercontent)
response["HX-Push-Url"] = reverse('events:rsvp-base', args=[rsvp.edit_token, page])
response["HX-Push-Url"] = reverse("events:rsvp-base", args=[rsvp.edit_token, page])
return response
# Email page
def _render_page_email(request, context, page) -> str:
note = summary_email(get_rsvp(context))
should_render = _render_page(request, get_rsvp(context), note, RSVP.Page.EMAIL, page)
if(type(should_render) is not bool):
return should_render
render_summary, summary = _render_page(
request, get_rsvp(context), note, RSVP.Page.EMAIL, page
)
if render_summary is True:
return summary
# Render the form
add_to_context(context, "form", RSVPForm(instance=get_rsvp(context)))
template = get_template("rsvp/rsvp_page_email.html")
return template.render(context=context, request=request)
def _render_page_guest(request, context, page) -> str:
note = summary_guests(get_rsvp(context))
should_render = _render_page(request, get_rsvp(context), note, RSVP.Page.GUESTS, page)
if(type(should_render) is not bool):
return should_render
guests = (
Guest.objects
.filter(rsvp=get_rsvp(context))
render_summary, summary = _render_page(
request, get_rsvp(context), note, RSVP.Page.GUESTS, page
)
add_to_context(context, "guests", guests)
if render_summary is True:
return summary
guests = Guest.objects.filter(rsvp=get_rsvp(context))
add_to_context(context, "guests", guests)
add_to_context(context, "form", GuestForm())
selected_activities = (
ActivitySelection.objects
.filter(rsvp=get_rsvp(context))
)
selected_activities = ActivitySelection.objects.filter(rsvp=get_rsvp(context))
# Check if any of the activities is full
selected_activities_full = [(True if act.activity.remaining_capacity() < 1 else False) for act in selected_activities]
add_to_context(context, "space_for_extra_guest", True not in selected_activities_full)
selected_activities_full = [
act.activity.remaining_capacity() < 1 for act in selected_activities
]
add_to_context(
context, "space_for_extra_guest", True not in selected_activities_full
)
template = get_template("rsvp/rsvp_page_guest.html")
return template.render(context=context, request=request)
def _render_page_activities(request, context, page) -> str:
note = summary_activities(get_rsvp(context))
should_render = _render_page(request, get_rsvp(context), note, RSVP.Page.ACTIVITIES, page)
if(type(should_render) is not bool):
return should_render
render_summary, summary = _render_page(
request, get_rsvp(context), note, RSVP.Page.ACTIVITIES, page
)
if render_summary is True:
return summary
activities = (
Activity.objects
.filter(event=get_event(context))
Activity.objects.filter(event=get_event(context))
.select_related("group")
.order_by("start_time")
)
groups = {}
ungrouped = []
for activity in activities:
if activity.group:
groups.setdefault(activity.group, []).append(activity)
else:
ungrouped.append(activity)
add_to_context(context, "groups", groups)
add_to_context(context, "ungrouped_activities", ungrouped)
selected_activities = (
ActivitySelection.objects
.filter(rsvp=get_rsvp(context))
)
add_to_context(context, "groups", groups)
add_to_context(context, "ungrouped_activities", ungrouped)
selected_activities = ActivitySelection.objects.filter(rsvp=get_rsvp(context))
selected_activities_ids = [act.activity.id for act in selected_activities]
add_to_context(context, "selected_activities", selected_activities_ids)
template = get_template("rsvp/rsvp_page_activities.html")
return template.render(context=context, request=request)
def _render_page_questions(request, context, page) -> str:
"""
Gather questions for activities selected by this RSVP and the guests list.
Pass `questions` and `guests` into the template context.
"""
note = summary_questions(get_rsvp(context))
should_render = _render_page(request, get_rsvp(context), note, RSVP.Page.QUESTIONS, page)
if(type(should_render) is not bool):
return should_render
render_summary, summary = _render_page(
request, get_rsvp(context), note, RSVP.Page.QUESTIONS, page
)
if render_summary is True:
return summary
rsvp = get_rsvp(context)
@@ -228,8 +245,7 @@ def _render_page_questions(request, context, page) -> str:
# Get all questions for the selected activities, with related activity and choices prefetched
if activities:
questions = (
Question.objects
.filter(activity__in=activities)
Question.objects.filter(activity__in=activities)
.select_related("activity")
.prefetch_related("choices")
.order_by("activity__start_time", "activity__title", "order")
@@ -242,9 +258,17 @@ def _render_page_questions(request, context, page) -> str:
# Existing responses so templates can prefill current answers.
# RSVP-scoped responses
responses_rsvp = Response.objects.filter(rsvp=rsvp, question__in=questions).select_related("question").prefetch_related("choices")
responses_rsvp = (
Response.objects.filter(rsvp=rsvp, question__in=questions)
.select_related("question")
.prefetch_related("choices")
)
# Guest-scoped responses (responses for guests of this rsvp)
responses_guest = Response.objects.filter(guest__rsvp=rsvp, question__in=questions).select_related("question", "guest").prefetch_related("choices")
responses_guest = (
Response.objects.filter(guest__rsvp=rsvp, question__in=questions)
.select_related("question", "guest")
.prefetch_related("choices")
)
add_to_context(context, "questions", questions)
add_to_context(context, "guests", guests)
@@ -254,88 +278,99 @@ def _render_page_questions(request, context, page) -> str:
template = get_template("rsvp/rsvp_page_questions.html")
return template.render(context=context, request=request)
def _render_page_notes(request, context, page) -> str:
note = summary_notes(get_rsvp(context))
should_render = _render_page(request, get_rsvp(context), note, RSVP.Page.NOTES, page)
if(type(should_render) is not bool):
return should_render
render_summary, summary = _render_page(
request, get_rsvp(context), note, RSVP.Page.NOTES, page
)
if render_summary is True:
return summary
add_to_context(context, "form", NotesForm(instance=get_rsvp(context)))
template = get_template("rsvp/rsvp_page_notes.html")
return template.render(context=context, request=request)
# Render generic page placeholder
def _render_page(request, rsvp:RSVP, note:str, page:RSVP.Page, current_page:RSVP.Page, oob:bool=False) -> bool | str:
def _render_page(
request,
rsvp: RSVP,
note: str,
page: RSVP.Page,
current_page: RSVP.Page,
oob: bool = False,
) -> tuple[bool, str]:
if page == current_page:
return True
return False, ""
template = get_template("rsvp/rsvp_page_inactive.html")
context = {
"note": note,
"rsvp": rsvp,
"page": page,
"complete": rsvp.is_page_finished(page),
"oob": oob
"oob": oob,
}
return template.render(context=context, request=request)
return True, template.render(context=context, request=request)
# Change the page of the request
def page_change(request, id:str, page:str):
logger.debug(f"page_change: id={id}, page={page}")
rsvp = get_object_or_404(RSVP, id=id)
def page_change(request, rid: str, page: str):
logger.debug("page_change: id=%s, page=%s", rid, page)
rsvp = get_object_or_404(RSVP, id=rid)
error = ""
if page == "next":
allowed = rsvp.is_next_page_allowed()
if(type(allowed) is bool and allowed == True):
allowed, error = rsvp.is_next_page_allowed()
if allowed is True:
rsvp.next_page()
elif type(allowed) is str:
# allowed contains an error
error = allowed
elif page == "previous":
rsvp.previous_page()
else:
rsvp.set_page(page)
# Done with rsvp
if rsvp.page == RSVP.Page.COMPLETED:
messages.success(request, "Thank you for responding!")
response = HttpResponse()
response["HX-Redirect"] = reverse('events:event', args=[rsvp.event.slug])
response["HX-Redirect"] = reverse("events:event", args=[rsvp.event.slug])
return response
return rsvp_pages(request, id, rsvp.page, error=error)
return rsvp_pages(request, rid, rsvp.page, error=error)
# Email step: handle post data
def post_step_email(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
if request.method != 'POST':
def post_step_email(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
if request.method != "POST":
return HttpResponseBadRequest()
form = RSVPForm(request.POST)
if form.is_valid():
rsvp.responder_email = form.cleaned_data['responder_email']
rsvp.send_email_updates = form.cleaned_data['send_email_updates']
rsvp.responder_email = form.cleaned_data["responder_email"]
rsvp.send_email_updates = form.cleaned_data["send_email_updates"]
rsvp.save()
return HttpResponse('Success', status=200)
return HttpResponse("Success", status=200)
return HttpResponseBadRequest(form.errors)
# Guest step: handle post data
def post_step_guest(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
if request.method != 'POST':
def post_step_guest(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
if request.method != "POST":
return HttpResponseBadRequest()
form = GuestForm(request.POST)
if form.is_valid():
guest = form.save(commit=False)
@@ -349,47 +384,55 @@ def post_step_guest(request, id:str):
return HttpResponseBadRequest(form.errors)
# Guest step: handle post data
def delete_step_guest(request, id:str, gid:str):
rsvp = get_object_or_404(RSVP, id=id)
def delete_step_guest(request, rid: str, gid: str):
rsvp = get_object_or_404(RSVP, id=rid)
guest = get_object_or_404(Guest, id=gid)
if request.method != 'DELETE':
if request.method != "DELETE":
return HttpResponseBadRequest()
if guest.rsvp.id != rsvp.id:
return HttpResponseBadRequest()
guest.delete()
content = _render_page_guest(request, create_context(rsvp), RSVP.Page.GUESTS)
return HttpResponse(content)
# Guest step: handle post data
def post_step_activities(request, id:str, aid:str):
rsvp = get_object_or_404(RSVP, id=id)
def post_step_activities(request, rid: str, aid: str):
rsvp = get_object_or_404(RSVP, id=rid)
activity = get_object_or_404(Activity, id=aid)
if request.method != 'POST':
if request.method != "POST":
return HttpResponseBadRequest()
try:
create_or_switch_activity_response(rsvp, activity)
except ValidationError as e:
logger.info(str(e))
return HttpResponseBadRequest(str(e))
content = _render_page_activities(request, create_context(rsvp), RSVP.Page.ACTIVITIES)
content = _render_page_activities(
request, create_context(rsvp), RSVP.Page.ACTIVITIES
)
# update the guests summary
note = summary_questions(rsvp)
summary = _render_page(request, rsvp, note, RSVP.Page.QUESTIONS, RSVP.Page.ACTIVITIES, oob=True)
render_summary, summary = _render_page(
request, rsvp, note, RSVP.Page.QUESTIONS, RSVP.Page.ACTIVITIES, oob=True
)
content = content + str(summary)
return HttpResponse(content)
def _create_or_update_response(response_kwargs, value_text=None, choice_obj=None):
def _create_or_update_response(
response_kwargs, value_text=None, choice_obj=None
) -> None:
"""
Ensure a single Response exists for (activity, question, rsvp|guest).
- For non-choice questions: store `value_text` and clear any choices.
@@ -398,10 +441,10 @@ def _create_or_update_response(response_kwargs, value_text=None, choice_obj=None
"""
# Find existing response for this owner/question/activity
existing = Response.objects.filter(
activity=response_kwargs['activity'],
question=response_kwargs['question'],
rsvp=response_kwargs.get('rsvp'),
guest=response_kwargs.get('guest'),
activity=response_kwargs["activity"],
question=response_kwargs["question"],
rsvp=response_kwargs.get("rsvp"),
guest=response_kwargs.get("guest"),
).first()
# Create response if it doesn't exist
@@ -426,61 +469,74 @@ def _create_or_update_response(response_kwargs, value_text=None, choice_obj=None
# If a choice was supplied, attach it via ResponseChoice
if choice_obj:
# question kind influences behavior: single should replace, multiple should add
kind = response_kwargs['question'].kind if 'question' in response_kwargs else None
kind = (
response_kwargs["question"].kind if "question" in response_kwargs else None
)
if kind == "single":
# Keep exactly this single choice for the response
ResponseChoice.objects.filter(response=resp).exclude(choice=choice_obj).delete()
ResponseChoice.objects.filter(response=resp).exclude(
choice=choice_obj
).delete()
ResponseChoice.objects.get_or_create(response=resp, choice=choice_obj)
else:
# multiple: ensure this choice exists (caller may call repeatedly for several choices)
ResponseChoice.objects.get_or_create(response=resp, choice=choice_obj)
return resp
# Question step: handle post data
def post_response_rsvp(request, id:str, qid:str):
def post_response_rsvp(request, rid: str, qid: str):
if request.method != "POST":
return HttpResponseBadRequest()
rsvp = get_object_or_404(RSVP, id=id)
rsvp = get_object_or_404(RSVP, id=rid)
question = get_object_or_404(Question, id=qid)
activity = question.activity
# Parse inputs:
if question.kind == "single":
choice_id = request.POST.get("choice")
choice = get_object_or_404(QuestionChoice, id=choice_id) if choice_id else None
resp = _create_or_update_response(
_create_or_update_response(
{"activity": activity, "question": question, "rsvp": rsvp},
choice_obj=choice
choice_obj=choice,
)
elif question.kind == "multiple":
choice_ids = request.POST.getlist("choices")
# For simplicity: create one Response per choice (or replace existing)
# Implementation detail depends on Response model use-case.
# (Example: delete existing and create new ones)
Response.objects.filter(activity=activity, question=question, rsvp=rsvp).delete()
Response.objects.filter(
activity=activity, question=question, rsvp=rsvp
).delete()
for cid in choice_ids:
choice = get_object_or_404(QuestionChoice, id=cid)
_create_or_update_response({"activity": activity, "question": question, "rsvp": rsvp}, choice_obj=choice)
_create_or_update_response(
{"activity": activity, "question": question, "rsvp": rsvp},
choice_obj=choice,
)
elif question.kind == "boolean":
val = request.POST.get("value_bool")
_create_or_update_response({"activity": activity, "question": question, "rsvp": rsvp}, value_text=val)
_create_or_update_response(
{"activity": activity, "question": question, "rsvp": rsvp}, value_text=val
)
else:
val = request.POST.get("value", "")
_create_or_update_response({"activity": activity, "question": question, "rsvp": rsvp}, value_text=val)
_create_or_update_response(
{"activity": activity, "question": question, "rsvp": rsvp}, value_text=val
)
# Re-render the questions page (or the single question partial). For simplicity render the whole questions page:
content = _render_page_questions(request, create_context(rsvp), RSVP.Page.QUESTIONS)
return HttpResponse(content)
def post_response_guest(request, id:str, qid:str, gid:str):
def post_response_guest(request, rid: str, qid: str, gid: str):
if request.method != "POST":
return HttpResponseBadRequest()
rsvp = get_object_or_404(RSVP, id=id)
rsvp = get_object_or_404(RSVP, id=rid)
question = get_object_or_404(Question, id=qid)
guest = get_object_or_404(Guest, id=gid)
activity = question.activity
@@ -491,36 +547,49 @@ def post_response_guest(request, id:str, qid:str, gid:str):
if question.kind == "single":
choice_id = request.POST.get("choice")
choice = get_object_or_404(QuestionChoice, id=choice_id) if choice_id else None
_create_or_update_response({"activity": activity, "question": question, "guest": guest}, choice_obj=choice)
_create_or_update_response(
{"activity": activity, "question": question, "guest": guest},
choice_obj=choice,
)
elif question.kind == "multiple":
choice_ids = request.POST.getlist("choices")
Response.objects.filter(activity=activity, question=question, guest=guest).delete()
Response.objects.filter(
activity=activity, question=question, guest=guest
).delete()
for cid in choice_ids:
choice = get_object_or_404(QuestionChoice, id=cid)
_create_or_update_response({"activity": activity, "question": question, "guest": guest}, choice_obj=choice)
_create_or_update_response(
{"activity": activity, "question": question, "guest": guest},
choice_obj=choice,
)
elif question.kind == "boolean":
val = request.POST.get("value_bool")
_create_or_update_response({"activity": activity, "question": question, "guest": guest}, value_text=val)
_create_or_update_response(
{"activity": activity, "question": question, "guest": guest}, value_text=val
)
else:
val = request.POST.get("value", "")
_create_or_update_response({"activity": activity, "question": question, "guest": guest}, value_text=val)
_create_or_update_response(
{"activity": activity, "question": question, "guest": guest}, value_text=val
)
content = _render_page_questions(request, create_context(rsvp), RSVP.Page.QUESTIONS)
return HttpResponse(content)
# Guest step: handle post data
def post_step_notes(request, id:str):
rsvp = get_object_or_404(RSVP, id=id)
if request.method != 'POST':
def post_step_notes(request, rid: str):
rsvp = get_object_or_404(RSVP, id=rid)
if request.method != "POST":
return HttpResponseBadRequest()
form = NotesForm(request.POST)
if form.is_valid():
rsvp.notes = form.cleaned_data['notes']
rsvp.notes = form.cleaned_data["notes"]
rsvp.save()
content = _render_page_notes(request, create_context(rsvp), RSVP.Page.NOTES)
return HttpResponse(content)
return HttpResponseBadRequest(form.errors)
return HttpResponseBadRequest(form.errors)
events/views/views.py
+33 -32
View File
@@ -1,5 +1,4 @@
# views.py
from email import message
import logging
from django.shortcuts import get_object_or_404, redirect, render
@@ -18,76 +17,78 @@ from ..models import (
# Logger
logger = logging.getLogger(__name__)
def invite_redirect(request:HttpRequest, slug:str):
def invite_redirect(request: HttpRequest, slug: str):
"""
Redirect from the unique invite page
to the rsvp response page which is linked to the response
"""
rsvp = get_object_or_404(RSVP, invite=slug)
response = redirect(reverse('events:event', args=[rsvp.event.slug]))
response = redirect(reverse("events:event", args=[rsvp.event.slug]))
# Opened for the first time
#TODO what is email not sent out?
if(rsvp.status == RSVP.Status.UNOPENED):
# TODO what if email not sent out?
if rsvp.status == RSVP.Status.UNOPENED:
rsvp.status = RSVP.Status.NOT_STARTED
rsvp.save()
# Only set the cookie the first time, after this they need to use the invite link
set_rsvp_cookie(response, rsvp.event, rsvp)
else:
messages.warning(request, "You can only use the invite once, please use the link in the email.")
logger.info(f"Invite `{slug}` opened for Event `{rsvp.event.title}`")
messages.warning(
request,
"You can only use the invite once, please use the link in the email.",
)
logger.info("Invite `%s` opened for Event `%s`", slug, rsvp.event.title)
return response
def event(request:HttpRequest, slug):
def event_page(request: HttpRequest, slug):
event = get_object_or_404(Event, slug=slug)
activities = (
Activity.objects
.filter(event=event)
Activity.objects.filter(event=event)
.select_related("group")
.order_by("start_time")
)
groups = {}
ungrouped = []
for activity in activities:
if activity.group:
groups.setdefault(activity.group, []).append(activity)
else:
ungrouped.append(activity)
context = {
"event": event,
"groups": groups,
"ungrouped_activities": ungrouped,
}
response = False
rsvp = get_rsvp_from_cookie(request, slug)
if type(rsvp) is RSVP:
joining = True if rsvp.status == RSVP.Status.YES_JOINING else False
not_coming = True if rsvp.status == RSVP.Status.NOT_COMING else False
in_progress = True if rsvp.status == RSVP.Status.INCOMPLETE else False
not_started = True if rsvp.status == RSVP.Status.NOT_STARTED or rsvp.status == RSVP.Status.UNOPENED else False
if isinstance(rsvp, RSVP):
joining = rsvp.status == RSVP.Status.YES_JOINING
not_coming = rsvp.status == RSVP.Status.NOT_COMING
in_progress = rsvp.status == RSVP.Status.INCOMPLETE
not_started = rsvp.status in (RSVP.Status.NOT_STARTED, RSVP.Status.UNOPENED)
response = {
"rsvp": rsvp,
"joining": joining,
"not_coming": not_coming,
"in_progress": in_progress,
"not_started": not_started
"not_started": not_started,
}
context["response"] = response
return render(request, "event/event.html", context)
return render(request, "event/event.html", context)
rsvpproject/asgi.py
+9 -8
View File
@@ -15,31 +15,32 @@ from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from blacknoise import BlackNoise
from django.core.asgi import get_asgi_application
logger = logging.getLogger(__file__)
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'rsvpproject.settings')
from django.conf import settings
from django.core.asgi import get_asgi_application
from events.routing import websocket_urlpatterns
logger = logging.getLogger(__file__)
if settings.DEBUG:
logger.warning("DEBUG enabled")
if settings.DOCKER:
logger.warning("DOCKER enabled")
logger.info(f"ALLOWED_HOSTS={settings.ALLOWED_HOSTS}")
logger.info("ALLOWED_HOSTS=%s", settings.ALLOWED_HOSTS)
django_asgi_app = BlackNoise(get_asgi_application())
django_asgi_app.add(settings.STATIC_ROOT, "/static")
django_asgi_app.add(settings.MEDIA_ROOT, "/media")
from events.routing import websocket_urlpatterns
application = ProtocolTypeRouter({
"http": django_asgi_app,
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
),
})
})
rsvpproject/settings.py
+76 -80
View File
@@ -13,13 +13,16 @@ https://docs.djangoproject.com/en/6.0/ref/settings/
import os
from pathlib import Path
from django.contrib.messages import constants as message_constants
bool_map = {"true": True, "false": False}
def extract_bool(env_key, default:bool):
def extract_bool(env_key, default: bool):
retval = default
env_val = os.getenv(env_key)
if env_val:
# Try as int
try:
@@ -29,31 +32,34 @@ def extract_bool(env_key, default:bool):
except ValueError:
# Convert string to boolean using the dictionary
retval = bool_map.get(env_val.lower(), default)
return retval
def extract_comma_list(env_key, default=None):
if os.getenv(env_key):
return [item.strip() for item in os.getenv(env_key).split(',')] # pyright: ignore[reportOptionalMemberAccess]
else:
if default:
return [default]
else:
return []
return [
item.strip() for item in os.getenv(env_key).split(",")
] # pyright: ignore[reportOptionalMemberAccess]
if default:
return [default]
return []
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/6.0/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = os.getenv('SECRET_KEY', 'INSECURE_STANDARD_KEY_SET_IN_ENV')
SECRET_KEY = os.getenv("SECRET_KEY", "INSECURE_STANDARD_KEY_SET_IN_ENV")
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = extract_bool('DEBUG', False)
DOCKER = extract_bool('DOCKER', False)
DEBUG = extract_bool("DEBUG", False)
DOCKER = extract_bool("DOCKER", False)
ALLOWED_HOSTS = extract_comma_list('ALLOWED_HOSTS', '127.0.0.1')
CSRF_TRUSTED_ORIGINS = extract_comma_list('CSRF_TRUSTED_ORIGINS')
ALLOWED_HOSTS = extract_comma_list("ALLOWED_HOSTS", "127.0.0.1")
CSRF_TRUSTED_ORIGINS = extract_comma_list("CSRF_TRUSTED_ORIGINS")
BASE_DIR = Path(__file__).resolve().parent.parent
@@ -61,64 +67,60 @@ DOCKER_DIR = BASE_DIR
if DOCKER:
# We are running in docker
DOCKER_DIR = Path('/data')
DOCKER_DIR = Path("/data")
# Application definition
INSTALLED_APPS = [
'daphne',
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'events',
"daphne",
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
"events",
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
"django.middleware.security.SecurityMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"django.middleware.clickjacking.XFrameOptionsMiddleware",
]
ROOT_URLCONF = 'rsvpproject.urls'
ROOT_URLCONF = "rsvpproject.urls"
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
'django.template.context_processors.csrf',
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
"django.template.context_processors.csrf",
],
},
},
]
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
},
'simple':{
'format': '%(levelname)s %(message)s'
},
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {"format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"},
"simple": {"format": "%(levelname)s %(message)s"},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'standard',
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "standard",
},
# 'file': {
# 'level': 'DEBUG' if DEBUG else 'INFO',
@@ -127,54 +129,48 @@ LOGGING = {
# 'formatter': 'simple',
# },
},
'loggers': {
'events': {
'handlers': ['console'],
'level': 'DEBUG' if DEBUG else 'INFO',
'propagate': True,
"loggers": {
"events": {
"handlers": ["console"],
"level": "DEBUG" if DEBUG else "INFO",
"propagate": True,
},
},
}
ASGI_APPLICATION = 'rsvpproject.asgi.application'
ASGI_APPLICATION = "rsvpproject.asgi.application"
# Database
# https://docs.djangoproject.com/en/6.0/ref/settings/#databases
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': DOCKER_DIR / 'db.sqlite3',
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": DOCKER_DIR / "db.sqlite3",
}
}
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels.layers.InMemoryChannelLayer"
}
}
CHANNEL_LAYERS = {"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}}
# Password validation
# https://docs.djangoproject.com/en/6.0/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
},
]
from django.contrib.messages import constants as message_constants
MESSAGE_TAGS = {
message_constants.INFO: "bg-blue",
message_constants.SUCCESS: "bg-green",
@@ -185,9 +181,9 @@ MESSAGE_TAGS = {
# Internationalization
# https://docs.djangoproject.com/en/6.0/topics/i18n/
LANGUAGE_CODE = 'en-us'
LANGUAGE_CODE = "en-us"
TIME_ZONE = 'UTC'
TIME_ZONE = "UTC"
USE_I18N = True
@@ -197,8 +193,8 @@ USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/6.0/howto/static-files/
STATIC_URL = 'static/'
STATIC_ROOT = BASE_DIR / 'static'
STATIC_URL = "static/"
STATIC_ROOT = BASE_DIR / "static"
MEDIA_URL = 'media/'
MEDIA_ROOT = DOCKER_DIR / 'media'
MEDIA_URL = "media/"
MEDIA_ROOT = DOCKER_DIR / "media"