feat: google cron mixin (#11)

* feat: google cron mixins

* max 80 lines

* max 80 lines pt.2

* add get method signature

codeforlife/kurono/avatar_state.py

@@ -13,7 +13,12 @@ def create_avatar_state(avatar_state_json: Dict):
     if avatar_state_json.get("backpack"):
         # use namedtuple for artefacts to allow accessing fields by name
         avatar_state_dict["backpack"] = Backpack(
-            [namedtuple("Artefact", artefact.keys())(*artefact.values()) for artefact in avatar_state_json["backpack"]]
+            [
+                namedtuple("Artefact", artefact.keys())(*artefact.values())
+                for artefact in avatar_state_json["backpack"]
+            ]
         )
 
-    return namedtuple("AvatarState", avatar_state_dict.keys())(*avatar_state_dict.values())
+    return namedtuple("AvatarState", avatar_state_dict.keys())(
+        *avatar_state_dict.values()
+    )

codeforlife/kurono/backpack.py

@@ -21,4 +21,11 @@ def find(self, artefact_type: str) -> int:
         Returns:
             The index of the artefact or -1 if it doesn't exist.
         """
-        return next((i for i, artefact in enumerate(self) if artefact.type == artefact_type), -1)
+        return next(
+            (
+                i
+                for i, artefact in enumerate(self)
+                if artefact.type == artefact_type
+            ),
+            -1,
+        )

codeforlife/kurono/pathfinding.py

@@ -16,7 +16,9 @@ def __init__(self, parent=None, cell=None):
         self.parent = parent
         self.cell = cell
 
-        self.g = 0  # g is the distance between the current node and the start node
+        self.g = (
+            0  # g is the distance between the current node and the start node
+        )
         self.h = 0  # h is the heuristic - estimated distance from the current node to the end node
         self.f = 0  # f is the total cost of the node (g + h)
 
@@ -107,13 +109,20 @@ def astar(world_map, start_cell, end_cell):
 
             # calculate the f, g, and h values
             child.g = current_node.g + 1
-            child.h = ((child.location.x - end_node.location.x) ** 2) + ((child.location.y - end_node.location.y) ** 2)
+            child.h = ((child.location.x - end_node.location.x) ** 2) + (
+                (child.location.y - end_node.location.y) ** 2
+            )
             child.f = child.g + child.h
 
             # check if it is already in the open list, and if this path to that square is better,
             # using G cost as the measure (lower G is better)
             open_nodes = [
-                open_node for open_node in open_list if (child.location == open_node.location and child.g > open_node.g)
+                open_node
+                for open_node in open_list
+                if (
+                    child.location == open_node.location
+                    and child.g > open_node.g
+                )
             ]
             if len(open_nodes) > 0:
                 continue

codeforlife/kurono/utils.py

@@ -16,6 +16,8 @@ def __getitem__(self, i):
             return super().__getitem__(i)
         except IndexError:
             if len(self) == 0:
-                print("There aren't any nearby artefacts, you need to move closer!")
+                print(
+                    "There aren't any nearby artefacts, you need to move closer!"
+                )
             else:
                 raise

codeforlife/kurono/world_map.py

@@ -54,10 +54,15 @@ def habitable(self):
         return not (self.avatar or self.obstacle)
 
     def has_artefact(self):
-        return self.interactable is not None and self.interactable["type"] in ARTEFACT_TYPES
+        return (
+            self.interactable is not None
+            and self.interactable["type"] in ARTEFACT_TYPES
+        )
 
     def __repr__(self):
-        return "Cell({} a={} i={})".format(self.location, self.avatar, self.interactable)
+        return "Cell({} a={} i={})".format(
+            self.location, self.avatar, self.interactable
+        )
 
     def __eq__(self, other):
         return self.location == other.location
@@ -76,7 +81,10 @@ def generate_world_map_from_cells_data(cells: List[Cell]) -> "WorldMap":
 
     def generate_world_map_from_game_state(game_state) -> "WorldMap":
         cells: Dict[Location, Cell] = {}
-        for x in range(game_state["southWestCorner"]["x"], game_state["northEastCorner"]["x"] + 1):
+        for x in range(
+            game_state["southWestCorner"]["x"],
+            game_state["northEastCorner"]["x"] + 1,
+        ):
             for y in range(
                 game_state["southWestCorner"]["y"],
                 game_state["northEastCorner"]["y"] + 1,
@@ -85,15 +93,21 @@ def generate_world_map_from_game_state(game_state) -> "WorldMap":
                 cells[Location(x, y)] = cell
 
         for interactable in game_state["interactables"]:
-            location = Location(interactable["location"]["x"], interactable["location"]["y"])
+            location = Location(
+                interactable["location"]["x"], interactable["location"]["y"]
+            )
             cells[location].interactable = interactable
 
         for obstacle in game_state["obstacles"]:
-            location = Location(obstacle["location"]["x"], obstacle["location"]["y"])
+            location = Location(
+                obstacle["location"]["x"], obstacle["location"]["y"]
+            )
             cells[location].obstacle = obstacle
 
         for player in game_state["players"]:
-            location = Location(player["location"]["x"], player["location"]["y"])
+            location = Location(
+                player["location"]["x"], player["location"]["y"]
+            )
             cells[location].player = create_avatar_state(player)
 
         return WorldMap(cells)
@@ -117,10 +131,18 @@ def interactable_cells(self):
         return [cell for cell in self.all_cells() if cell.interactable]
 
     def pickup_cells(self):
-        return [cell for cell in self.interactable_cells() if cell.interactable["type"] in PICKUP_TYPES]
+        return [
+            cell
+            for cell in self.interactable_cells()
+            if cell.interactable["type"] in PICKUP_TYPES
+        ]
 
     def score_cells(self):
-        return [cell for cell in self.interactable_cells() if "score" == cell.interactable["type"]]
+        return [
+            cell
+            for cell in self.interactable_cells()
+            if "score" == cell.interactable["type"]
+        ]
 
     def partially_fogged_cells(self):
         return [cell for cell in self.all_cells() if cell.partially_fogged]
@@ -130,21 +152,31 @@ def is_visible(self, location):
 
     def get_cell(self, location):
         cell = self.cells[location]
-        assert cell.location == location, "location lookup mismatch: arg={}, found={}".format(location, cell.location)
+        assert (
+            cell.location == location
+        ), "location lookup mismatch: arg={}, found={}".format(
+            location, cell.location
+        )
         return cell
 
     def can_move_to(self, target_location):
         try:
             cell = self.get_cell(target_location)
         except KeyError:
             return False
-        return getattr(cell, "habitable", False) and not getattr(cell, "avatar", False)
+        return getattr(cell, "habitable", False) and not getattr(
+            cell, "avatar", False
+        )
 
     def _scan_artefacts(self, start_location, radius):
         # get artefacts from starting location within the radius
         artefacts = []
-        for x in range(start_location.x - radius, start_location.x + radius + 1):
-            for y in range(start_location.y - radius, start_location.y + radius + 1):
+        for x in range(
+            start_location.x - radius, start_location.x + radius + 1
+        ):
+            for y in range(
+                start_location.y - radius, start_location.y + radius + 1
+            ):
                 try:
                     cell = self.get_cell(Location(x, y))
                 except KeyError:
@@ -153,7 +185,9 @@ def _scan_artefacts(self, start_location, radius):
                     artefacts.append(cell)
         return artefacts
 
-    def scan_nearby(self, avatar_location, radius=SCAN_RADIUS) -> NearbyArtefactsList[dict]:
+    def scan_nearby(
+        self, avatar_location, radius=SCAN_RADIUS
+    ) -> NearbyArtefactsList[dict]:
         """
         From the given location point search the given radius for artefacts.
         Returns list of nearest artefacts (artefact/interactable represented as dict).

codeforlife/mixins/__init__.py

@@ -0,0 +1 @@
+from .cron_mixin import CronMixin

codeforlife/mixins/cron_mixin.py

@@ -0,0 +1,17 @@
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from ..permissions import IsCronRequestFromGoogle
+
+
+class CronMixin:
+    """
+    A cron job on Google's AppEngine.
+    https://cloud.google.com/appengine/docs/flexible/scheduling-jobs-with-cron-yaml
+    """
+
+    http_method_names = ["get"]
+    permission_classes = [IsCronRequestFromGoogle]
+
+    def get(self, request: Request) -> Response:
+        raise NotImplementedError()

codeforlife/permissions/__init__.py

@@ -0,0 +1 @@
+from .is_cron_request_from_google import IsCronRequestFromGoogle

codeforlife/permissions/is_cron_request_from_google.py

@@ -0,0 +1,18 @@
+from django.conf import settings
+from rest_framework.permissions import BasePermission
+from rest_framework.request import Request
+from rest_framework.views import View
+
+
+class IsCronRequestFromGoogle(BasePermission):
+    """
+    Validate that requests to your cron URLs are coming from App Engine and not
+    from another source.
+    https://cloud.google.com/appengine/docs/flexible/scheduling-jobs-with-cron-yaml#securing_urls_for_cron
+    """
+
+    def has_permission(self, request: Request, view: View):
+        return (
+            settings.DEBUG
+            or request.META.get("HTTP_X_APPENGINE_CRON") == "true"
+        )

codeforlife/user/admin.py

@@ -20,14 +20,35 @@ class CustomUserAdmin(UserAdmin):
 
     fieldsets = (
         (None, {"fields": ("username", "email", "password")}),
-        ("Permissions", {"fields": ("is_staff", "is_active", "is_superuser", "groups", "user_permissions")}),
+        (
+            "Permissions",
+            {
+                "fields": (
+                    "is_staff",
+                    "is_active",
+                    "is_superuser",
+                    "groups",
+                    "user_permissions",
+                )
+            },
+        ),
         ("Dates", {"fields": ("last_login", "date_joined")}),
     )
 
     add_fieldsets = (
         (
             None,
-            {"classes": ("wide",), "fields": ("username", "email", "password1", "password2", "is_staff", "is_active")},
+            {
+                "classes": ("wide",),
+                "fields": (
+                    "username",
+                    "email",
+                    "password1",
+                    "password2",
+                    "is_staff",
+                    "is_active",
+                ),
+            },
         ),
     )
 

codeforlife/user/models/classroom.py

@@ -29,15 +29,21 @@ def get_queryset(self):
 
 class Class(models.Model):
     name = models.CharField(max_length=200)
-    teacher = models.ForeignKey(Teacher, related_name="class_teacher", on_delete=models.CASCADE)
+    teacher = models.ForeignKey(
+        Teacher, related_name="class_teacher", on_delete=models.CASCADE
+    )
     access_code = models.CharField(max_length=5, null=True)
     classmates_data_viewable = models.BooleanField(default=False)
     always_accept_requests = models.BooleanField(default=False)
     accept_requests_until = models.DateTimeField(null=True)
     creation_time = models.DateTimeField(default=timezone.now, null=True)
     is_active = models.BooleanField(default=True)
     created_by = models.ForeignKey(
-        Teacher, null=True, blank=True, related_name="created_classes", on_delete=models.SET_NULL
+        Teacher,
+        null=True,
+        blank=True,
+        related_name="created_classes",
+        on_delete=models.SET_NULL,
     )
 
     objects = ClassModelManager()
@@ -49,7 +55,9 @@ def __str__(self):
     def active_game(self):
         games = self.game_set.filter(game_class=self, is_archived=False)
         if len(games) >= 1:
-            assert len(games) == 1  # there should NOT be more than one active game
+            assert (
+                len(games) == 1
+            )  # there should NOT be more than one active game
             return games[0]
         return None
 
@@ -59,16 +67,23 @@ def has_students(self):
 
     def get_requests_message(self):
         if self.always_accept_requests:
-            external_requests_message = "This class is currently set to always accept requests."
-        elif self.accept_requests_until is not None and (self.accept_requests_until - timezone.now()) >= timedelta():
+            external_requests_message = (
+                "This class is currently set to always accept requests."
+            )
+        elif (
+            self.accept_requests_until is not None
+            and (self.accept_requests_until - timezone.now()) >= timedelta()
+        ):
             external_requests_message = (
                 "This class is accepting external requests until "
                 + self.accept_requests_until.strftime("%d-%m-%Y %H:%M")
                 + " "
                 + timezone.get_current_timezone_name()
             )
         else:
-            external_requests_message = "This class is not currently accepting external requests."
+            external_requests_message = (
+                "This class is not currently accepting external requests."
+            )
 
         return external_requests_message
 

codeforlife/user/models/other.py

@@ -19,7 +19,9 @@ class JoinReleaseStudent(models.Model):
     JOIN = "join"
     RELEASE = "release"
 
-    student = models.ForeignKey(Student, related_name="student", on_delete=models.CASCADE)
+    student = models.ForeignKey(
+        Student, related_name="student", on_delete=models.CASCADE
+    )
     # either "release" or "join"
     action_type = models.CharField(max_length=64)
     action_time = models.DateTimeField(default=timezone.now)

codeforlife/user/models/school.py

@@ -35,7 +35,11 @@ def classes(self):
 
     def admins(self):
         teachers = self.school_teacher.all()
-        return [teacher for teacher in teachers if teacher.is_admin] if teachers else None
+        return (
+            [teacher for teacher in teachers if teacher.is_admin]
+            if teachers
+            else None
+        )
 
     def anonymise(self):
         self.name = uuid4().hex

codeforlife/user/models/session.py

@@ -11,7 +11,9 @@ class UserSession(models.Model):
     login_time = models.DateTimeField(default=timezone.now)
     school = models.ForeignKey(School, null=True, on_delete=models.SET_NULL)
     class_field = models.ForeignKey(Class, null=True, on_delete=models.SET_NULL)
-    login_type = models.CharField(max_length=100, null=True)  # for student login
+    login_type = models.CharField(
+        max_length=100, null=True
+    )  # for student login
 
     def __str__(self):
         return f"{self.user} login: {self.login_time} type: {self.login_type}"