chore: fix pghistory migration deps and improve htmx utilities

- Update pghistory dependency from 0007 to 0006 in account migrations
- Add docstrings and remove unused imports in htmx_forms.py
- Add DJANGO_SETTINGS_MODULE bash commands to Claude settings
- Add state transition definitions for ride statuses
This commit is contained in:
pacnpal
2025-12-21 17:33:24 -05:00
parent b9063ff4f8
commit 7ba0004c93
74 changed files with 11134 additions and 198 deletions

View File

@@ -0,0 +1,211 @@
"""
Services for ride status transitions and management.
Following Django styleguide pattern for business logic encapsulation.
"""
from typing import Optional
from django.db import transaction
from django.contrib.auth.models import AbstractBaseUser
from apps.rides.models import Ride
class RideStatusService:
"""Service for managing ride status transitions using FSM."""
@staticmethod
def open_ride(*, ride_id: int, user: Optional[AbstractBaseUser] = None) -> Ride:
"""
Open a ride for operation.
Args:
ride_id: ID of ride to open
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.open(user=user)
return ride
@staticmethod
def close_ride_temporarily(
*, ride_id: int, user: Optional[AbstractBaseUser] = None
) -> Ride:
"""
Temporarily close a ride.
Args:
ride_id: ID of ride to close temporarily
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.close_temporarily(user=user)
return ride
@staticmethod
def mark_ride_sbno(
*, ride_id: int, user: Optional[AbstractBaseUser] = None
) -> Ride:
"""
Mark a ride as SBNO (Standing But Not Operating).
Args:
ride_id: ID of ride to mark as SBNO
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.mark_sbno(user=user)
return ride
@staticmethod
def mark_ride_closing(
*,
ride_id: int,
closing_date,
post_closing_status: str,
user: Optional[AbstractBaseUser] = None,
) -> Ride:
"""
Mark a ride as closing with a specific date and post-closing status.
Args:
ride_id: ID of ride to mark as closing
closing_date: Date when ride will close
post_closing_status: Status to transition to after closing
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
ValidationError: If post_closing_status is not set
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.mark_closing(
closing_date=closing_date,
post_closing_status=post_closing_status,
user=user,
)
return ride
@staticmethod
def close_ride_permanently(
*, ride_id: int, user: Optional[AbstractBaseUser] = None
) -> Ride:
"""
Permanently close a ride.
Args:
ride_id: ID of ride to close permanently
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.close_permanently(user=user)
return ride
@staticmethod
def demolish_ride(*, ride_id: int, user: Optional[AbstractBaseUser] = None) -> Ride:
"""
Mark a ride as demolished.
Args:
ride_id: ID of ride to demolish
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.demolish(user=user)
return ride
@staticmethod
def relocate_ride(*, ride_id: int, user: Optional[AbstractBaseUser] = None) -> Ride:
"""
Mark a ride as relocated.
Args:
ride_id: ID of ride to relocate
user: User performing the action
Returns:
Updated Ride instance
Raises:
Ride.DoesNotExist: If ride doesn't exist
"""
with transaction.atomic():
ride = Ride.objects.select_for_update().get(id=ride_id)
ride.relocate(user=user)
return ride
@staticmethod
def process_closing_rides() -> list[Ride]:
"""
Process all rides in CLOSING status and transition them to their
post_closing_status if the closing_date has been reached.
Returns:
List of rides that were transitioned
Note:
This method should be called by a scheduled task/cron job.
"""
from django.utils import timezone
transitioned_rides = []
closing_rides = Ride.objects.filter(
status="CLOSING",
closing_date__lte=timezone.now().date(),
).select_for_update()
for ride in closing_rides:
try:
with transaction.atomic():
ride.apply_post_closing_status()
transitioned_rides.append(ride)
except Exception as e:
# Log error but continue processing other rides
import logging
logger = logging.getLogger(__name__)
logger.error(
f"Failed to process closing ride {ride.id}: {e}",
exc_info=True,
)
continue
return transitioned_rides