Add ClubMembership enhancements, CSV import command, and related tests
- Extend the ClubMembership model with `club`, `member`, and optional `license` fields, along with relevant constraints and ordering. - Implement verbose names for Club and ClubMembership models and update admin configurations for better display and filtering. - Add a `import_members_csv` management command for batch importing members, clubs, and memberships from a CSV file. - Include extensive tests for the `import_members_csv` command, ClubMembership model, and Club model. - Refactor related migrations, services, and test structure.
This commit is contained in:
0
authentication/management/__init__.py
Normal file
0
authentication/management/__init__.py
Normal file
0
authentication/management/commands/__init__.py
Normal file
0
authentication/management/commands/__init__.py
Normal file
43
authentication/management/commands/import_members_csv.py
Normal file
43
authentication/management/commands/import_members_csv.py
Normal file
@@ -0,0 +1,43 @@
|
||||
from django.core.management.base import BaseCommand, CommandError
|
||||
|
||||
from authentication.services.member_csv_importer import MemberCsvImporter
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Import club members from a CSV file."
|
||||
|
||||
def add_arguments(self, parser) -> None:
|
||||
parser.add_argument(
|
||||
"csv_file",
|
||||
type=str,
|
||||
help="Path to the CSV file to import.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--date-format",
|
||||
default="%Y-%m-%d",
|
||||
help="Date format for date_of_birth. Default: %%Y-%%m-%%d",
|
||||
)
|
||||
|
||||
def handle(self, *args, **options) -> str | None:
|
||||
importer = MemberCsvImporter(date_format=options["date_format"])
|
||||
|
||||
try:
|
||||
result = importer.import_path(options["csv_file"])
|
||||
except Exception as exc:
|
||||
raise CommandError(str(exc)) from exc
|
||||
|
||||
for error in result.errors:
|
||||
self.stderr.write(self.style.ERROR(f"Row {error.row_number} skipped: {error.message}"))
|
||||
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(
|
||||
"Import complete. "
|
||||
f"Members created: {result.created_members}. "
|
||||
f"Members updated: {result.updated_members}. "
|
||||
f"Users created: {result.created_users}. "
|
||||
f"Clubs created: {result.created_clubs}. "
|
||||
f"Memberships created: {result.created_memberships}. "
|
||||
f"Memberships updated: {result.updated_memberships}. "
|
||||
f"Rows skipped: {result.skipped_rows}."
|
||||
)
|
||||
)
|
||||
0
authentication/services/__init__.py
Normal file
0
authentication/services/__init__.py
Normal file
180
authentication/services/member_csv_importer.py
Normal file
180
authentication/services/member_csv_importer.py
Normal file
@@ -0,0 +1,180 @@
|
||||
import csv
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import IO, Any
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.db import transaction
|
||||
|
||||
from authentication.models import Member
|
||||
from club.models import Club, ClubMembership
|
||||
|
||||
REQUIRED_COLUMNS = {"first_name", "last_name", "email", "date_of_birth", "create_account", "club_name", "license_number"}
|
||||
TRUE_VALUES = {"1", "true", "yes", "y", "on"}
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
@dataclass
|
||||
class MemberImportRowError:
|
||||
row_number: int
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class MemberImportResult:
|
||||
created_members: int = 0
|
||||
updated_members: int = 0
|
||||
created_users: int = 0
|
||||
created_clubs: int = 0
|
||||
created_memberships: int = 0
|
||||
updated_memberships: int = 0
|
||||
skipped_rows: int = 0
|
||||
errors: list[MemberImportRowError] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def successful_rows(self):
|
||||
return self.created_members + self.updated_members
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImportedMemberRowResult:
|
||||
member_created: bool
|
||||
user_created: bool
|
||||
club_created: bool
|
||||
membership_created: bool
|
||||
|
||||
|
||||
class MemberCsvImporter:
|
||||
def __init__(self, *, date_format="%Y-%m-%d") -> None:
|
||||
self.date_format = date_format
|
||||
|
||||
def import_path(self, csv_path) -> MemberImportResult:
|
||||
path = Path(csv_path)
|
||||
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"CSV file does not exist: {path}")
|
||||
|
||||
with path.open(newline="", encoding="utf-8-sig") as csv_file:
|
||||
return self.import_file(csv_file)
|
||||
|
||||
def import_file(self, csv_file: IO[str]) -> MemberImportResult:
|
||||
result = MemberImportResult()
|
||||
reader = csv.DictReader(csv_file)
|
||||
|
||||
if reader.fieldnames is None:
|
||||
raise ValueError("CSV file is empty or missing a header row.")
|
||||
|
||||
missing_columns = REQUIRED_COLUMNS - set(reader.fieldnames)
|
||||
if missing_columns:
|
||||
raise ValueError("CSV file is missing required columns: " + ", ".join(sorted(missing_columns)))
|
||||
|
||||
for row_number, row in enumerate(reader, start=2):
|
||||
try:
|
||||
row_result = self.import_row(row)
|
||||
except Exception as exc:
|
||||
result.skipped_rows += 1
|
||||
result.errors.append(
|
||||
MemberImportRowError(
|
||||
row_number=row_number,
|
||||
message=str(exc),
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
if row_result.member_created:
|
||||
result.created_members += 1
|
||||
else:
|
||||
result.updated_members += 1
|
||||
|
||||
if row_result.user_created:
|
||||
result.created_users += 1
|
||||
|
||||
if row_result.club_created:
|
||||
result.created_clubs += 1
|
||||
|
||||
if row_result.membership_created:
|
||||
result.created_memberships += 1
|
||||
else:
|
||||
result.updated_memberships += 1
|
||||
|
||||
return result
|
||||
|
||||
@transaction.atomic
|
||||
def import_row(self, row) -> ImportedMemberRowResult:
|
||||
first_name = self.clean_required(row, "first_name")
|
||||
last_name = self.clean_required(row, "last_name")
|
||||
email = self.clean_required(row, "email").lower()
|
||||
date_of_birth = self.parse_date(self.clean_required(row, "date_of_birth"))
|
||||
create_account = self.as_bool(row.get("create_account", ""))
|
||||
club_name = self.clean_required(row, "club_name")
|
||||
license_number = row.get("license_number", "").strip()
|
||||
|
||||
user = None
|
||||
user_created = False
|
||||
|
||||
if create_account:
|
||||
user, user_created = self.get_or_create_user(email)
|
||||
|
||||
member, member_created = Member.objects.update_or_create(
|
||||
email=email,
|
||||
defaults={
|
||||
"first_name": first_name,
|
||||
"last_name": last_name,
|
||||
"date_of_birth": date_of_birth,
|
||||
"user": user,
|
||||
},
|
||||
)
|
||||
|
||||
if create_account and member.user_id is None:
|
||||
member.user = user
|
||||
member.save(update_fields=["user"])
|
||||
|
||||
club, club_created = Club.objects.get_or_create(name=club_name)
|
||||
|
||||
_, membership_created = ClubMembership.objects.update_or_create(
|
||||
club=club,
|
||||
member=member,
|
||||
defaults={
|
||||
"license": license_number,
|
||||
},
|
||||
)
|
||||
|
||||
return ImportedMemberRowResult(
|
||||
member_created=member_created,
|
||||
user_created=user_created,
|
||||
club_created=club_created,
|
||||
membership_created=membership_created,
|
||||
)
|
||||
|
||||
def get_or_create_user(self, email) -> tuple[User, bool]:
|
||||
user, created = User.objects.get_or_create(
|
||||
email=email,
|
||||
defaults={
|
||||
"is_active": True,
|
||||
},
|
||||
)
|
||||
|
||||
if created:
|
||||
user.set_unusable_password()
|
||||
user.save(update_fields=["password"])
|
||||
|
||||
return user, created
|
||||
|
||||
def clean_required(self, row, field_name) -> Any:
|
||||
value = row.get(field_name, "").strip()
|
||||
|
||||
if not value:
|
||||
raise ValueError(f"{field_name} is required.")
|
||||
|
||||
return value
|
||||
|
||||
def parse_date(self, value) -> date:
|
||||
try:
|
||||
return datetime.strptime(value, self.date_format).date()
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"Invalid date_of_birth '{value}'. Expected format: {self.date_format}.") from exc
|
||||
|
||||
def as_bool(self, value) -> bool:
|
||||
return value.strip().lower() in TRUE_VALUES
|
||||
@@ -1,11 +1,18 @@
|
||||
import tempfile
|
||||
import uuid
|
||||
from datetime import date
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.db import IntegrityError
|
||||
from django.db.models import SET_NULL
|
||||
from django.test import TestCase
|
||||
|
||||
from authentication.models import Family, FamilyMembership, Member
|
||||
from club.models import Club, ClubMembership
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
@@ -332,3 +339,233 @@ class AdminSmokeTests(TestCase):
|
||||
{"app_label": "authentication", "model_name": "member", "field_name": "user", "term": "root"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
|
||||
class ImportMembersCsvCommandTests(TestCase):
|
||||
def write_csv(self, content):
|
||||
temp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False, encoding="utf-8")
|
||||
temp_file.write(content)
|
||||
temp_file.close()
|
||||
self.addCleanup(lambda: Path(temp_file.name).unlink(missing_ok=True))
|
||||
return temp_file.name
|
||||
|
||||
def call_import_command(self, csv_path, **options):
|
||||
stdout = StringIO()
|
||||
stderr = StringIO()
|
||||
|
||||
call_command(
|
||||
"import_members_csv",
|
||||
csv_path,
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
**options,
|
||||
)
|
||||
|
||||
return stdout.getvalue(), stderr.getvalue()
|
||||
|
||||
def test_import_creates_member_club_membership_and_user_when_requested(self):
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email,date_of_birth,create_account,club_name,license_number",
|
||||
"Jane,Doe,jane@example.com,2010-04-12,true,City Swim Club,LIC-001",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout, stderr = self.call_import_command(csv_path)
|
||||
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertIn("Import complete.", stdout)
|
||||
self.assertIn("Members created: 1.", stdout)
|
||||
self.assertIn("Users created: 1.", stdout)
|
||||
self.assertIn("Clubs created: 1.", stdout)
|
||||
self.assertIn("Memberships created: 1.", stdout)
|
||||
self.assertIn("Rows skipped: 0.", stdout)
|
||||
|
||||
member = Member.objects.get(email="jane@example.com")
|
||||
self.assertEqual(member.first_name, "Jane")
|
||||
self.assertEqual(member.last_name, "Doe")
|
||||
self.assertEqual(member.date_of_birth, date(2010, 4, 12))
|
||||
self.assertIsNotNone(member.user)
|
||||
self.assertEqual(member.user.email, "jane@example.com")
|
||||
self.assertFalse(member.user.has_usable_password())
|
||||
|
||||
club = Club.objects.get(name="City Swim Club")
|
||||
membership = ClubMembership.objects.get(club=club, member=member)
|
||||
self.assertEqual(membership.license, "LIC-001")
|
||||
|
||||
def test_import_creates_member_without_user_when_create_account_is_false(self):
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email,date_of_birth,create_account,club_name,license_number",
|
||||
"John,Smith,john@example.com,2009-11-03,false,City Swim Club,LIC-002",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout, stderr = self.call_import_command(csv_path)
|
||||
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertIn("Members created: 1.", stdout)
|
||||
self.assertIn("Users created: 0.", stdout)
|
||||
|
||||
member = Member.objects.get(email="john@example.com")
|
||||
self.assertIsNone(member.user)
|
||||
self.assertFalse(User.objects.filter(email="john@example.com").exists())
|
||||
|
||||
def test_import_updates_existing_member_and_membership(self):
|
||||
club = Club.objects.create(name="City Swim Club")
|
||||
member = Member.objects.create(
|
||||
first_name="Old",
|
||||
last_name="Name",
|
||||
email="jane@example.com",
|
||||
date_of_birth=date(2010, 1, 1),
|
||||
)
|
||||
ClubMembership.objects.create(
|
||||
club=club,
|
||||
member=member,
|
||||
license="OLD-LIC",
|
||||
)
|
||||
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email,date_of_birth,create_account,club_name,license_number",
|
||||
"Jane,Doe,jane@example.com,2010-04-12,false,City Swim Club,LIC-NEW",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout, stderr = self.call_import_command(csv_path)
|
||||
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertIn("Members created: 0.", stdout)
|
||||
self.assertIn("Members updated: 1.", stdout)
|
||||
self.assertIn("Memberships created: 0.", stdout)
|
||||
self.assertIn("Memberships updated: 1.", stdout)
|
||||
|
||||
member.refresh_from_db()
|
||||
self.assertEqual(member.first_name, "Jane")
|
||||
self.assertEqual(member.last_name, "Doe")
|
||||
self.assertEqual(member.date_of_birth, date(2010, 4, 12))
|
||||
|
||||
membership = ClubMembership.objects.get(club=club, member=member)
|
||||
self.assertEqual(membership.license, "LIC-NEW")
|
||||
|
||||
def test_import_links_existing_user_when_create_account_is_true(self):
|
||||
user = User.objects.create_user(email="jane@example.com", password="secret123")
|
||||
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email,date_of_birth,create_account,club_name,license_number",
|
||||
"Jane,Doe,jane@example.com,2010-04-12,true,City Swim Club,LIC-001",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout, stderr = self.call_import_command(csv_path)
|
||||
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertIn("Users created: 0.", stdout)
|
||||
|
||||
member = Member.objects.get(email="jane@example.com")
|
||||
self.assertEqual(member.user, user)
|
||||
self.assertTrue(user.check_password("secret123"))
|
||||
|
||||
def test_import_supports_custom_date_format(self):
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email,date_of_birth,create_account,club_name,license_number",
|
||||
"Jane,Doe,jane@example.com,12/04/2010,false,City Swim Club,LIC-001",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout, stderr = self.call_import_command(csv_path, date_format="%d/%m/%Y")
|
||||
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertIn("Members created: 1.", stdout)
|
||||
|
||||
member = Member.objects.get(email="jane@example.com")
|
||||
self.assertEqual(member.date_of_birth, date(2010, 4, 12))
|
||||
|
||||
def test_import_skips_invalid_row_and_imports_valid_rows(self):
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email,date_of_birth,create_account,club_name,license_number",
|
||||
"Jane,Doe,jane@example.com,2010-04-12,false,City Swim Club,LIC-001",
|
||||
"Broken,Date,broken@example.com,not-a-date,false,City Swim Club,LIC-002",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout, stderr = self.call_import_command(csv_path)
|
||||
|
||||
self.assertIn("Row 3 skipped:", stderr)
|
||||
self.assertIn("Invalid date_of_birth 'not-a-date'.", stderr)
|
||||
self.assertIn("Members created: 1.", stdout)
|
||||
self.assertIn("Rows skipped: 1.", stdout)
|
||||
|
||||
self.assertTrue(Member.objects.filter(email="jane@example.com").exists())
|
||||
self.assertFalse(Member.objects.filter(email="broken@example.com").exists())
|
||||
|
||||
def test_import_fails_for_missing_file(self):
|
||||
stdout = StringIO()
|
||||
stderr = StringIO()
|
||||
|
||||
with self.assertRaises(CommandError) as context:
|
||||
call_command(
|
||||
"import_members_csv",
|
||||
"does-not-exist.csv",
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
)
|
||||
|
||||
self.assertIn("CSV file does not exist", str(context.exception))
|
||||
|
||||
def test_import_fails_for_missing_required_columns(self):
|
||||
csv_path = self.write_csv(
|
||||
"\n".join(
|
||||
[
|
||||
"first_name,last_name,email",
|
||||
"Jane,Doe,jane@example.com",
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stdout = StringIO()
|
||||
stderr = StringIO()
|
||||
|
||||
with self.assertRaises(CommandError) as context:
|
||||
call_command(
|
||||
"import_members_csv",
|
||||
csv_path,
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
)
|
||||
|
||||
self.assertIn("CSV file is missing required columns:", str(context.exception))
|
||||
self.assertIn("club_name", str(context.exception))
|
||||
self.assertIn("date_of_birth", str(context.exception))
|
||||
self.assertIn("license_number", str(context.exception))
|
||||
|
||||
def test_import_fails_for_empty_csv_file(self):
|
||||
csv_path = self.write_csv("")
|
||||
|
||||
stdout = StringIO()
|
||||
stderr = StringIO()
|
||||
|
||||
with self.assertRaises(CommandError) as context:
|
||||
call_command(
|
||||
"import_members_csv",
|
||||
csv_path,
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
)
|
||||
|
||||
self.assertIn("CSV file is empty or missing a header row.", str(context.exception))
|
||||
|
||||
Reference in New Issue
Block a user