1from __future__ import annotations
2
3import re
4from datetime import datetime
5from typing import TYPE_CHECKING
6
7from plain import postgres
8from plain.postgres import types
9
10if TYPE_CHECKING:
11 from plain.http import Request
12
13__all__ = ["NotFoundLog", "Redirect", "RedirectLog"]
14
15
16@postgres.register_model
17class Redirect(postgres.Model):
18 from_pattern: str = types.TextField(max_length=255)
19 to_pattern: str = types.TextField(max_length=255)
20 http_status: int = types.SmallIntegerField(
21 default=301
22 ) # Default to permanent - could be choices?
23 created_at: datetime = types.DateTimeField(create_now=True)
24 updated_at: datetime = types.DateTimeField(create_now=True, update_now=True)
25 order: int = types.SmallIntegerField(default=0)
26 enabled: bool = types.BooleanField(default=True)
27 is_regex: bool = types.BooleanField(default=False)
28
29 # query params?
30 # logged in or not? auth not required necessarily...
31 # headers?
32
33 query: postgres.QuerySet[Redirect] = postgres.QuerySet()
34
35 model_options = postgres.Options(
36 ordering=["order", "-created_at"],
37 indexes=[
38 postgres.Index(
39 name="plainredirection_redirect_order_idx", fields=["order"]
40 ),
41 postgres.Index(
42 name="plainredirection_redirect_created_at_idx", fields=["created_at"]
43 ),
44 ],
45 constraints=[
46 postgres.UniqueConstraint(
47 fields=["from_pattern"],
48 name="plainredirects_redirect_unique_from_pattern",
49 ),
50 postgres.CheckConstraint(
51 check=postgres.Q(http_status__gte=0),
52 name="plainredirection_redirect_http_status_check",
53 ),
54 postgres.CheckConstraint(
55 check=postgres.Q(order__gte=0),
56 name="plainredirection_redirect_order_check",
57 ),
58 ],
59 )
60
61 def __str__(self) -> str:
62 return f"{self.from_pattern}"
63
64 def matches_request(self, request: Request) -> bool:
65 """
66 Decide whether a request matches this Redirect,
67 automatically checking whether the pattern is path based or full URL based.
68 """
69
70 if self.from_pattern.startswith("http"):
71 # Full url with query params
72 url = request.build_absolute_uri()
73 else:
74 # Doesn't include query params or host
75 url = request.path
76
77 if self.is_regex:
78 return bool(re.match(self.from_pattern, url))
79 else:
80 return url == self.from_pattern
81
82 def get_redirect_url(self, request: Request) -> str:
83 if not self.is_regex:
84 return self.to_pattern
85
86 # Replace any regex groups in the to_pattern
87 if self.from_pattern.startswith("http"):
88 url = request.build_absolute_uri()
89 else:
90 url = request.path
91
92 return re.sub(self.from_pattern, self.to_pattern, url)
93
94
95@postgres.register_model
96class RedirectLog(postgres.Model):
97 redirect: Redirect = types.ForeignKeyField(Redirect, on_delete=postgres.CASCADE)
98
99 # The actuals that were used to redirect
100 from_url: str = types.URLField(max_length=512)
101 to_url: str = types.URLField(max_length=512)
102 http_status: int = types.SmallIntegerField(default=301)
103
104 # Request metadata
105 ip_address: str = types.GenericIPAddressField()
106 user_agent: str = types.TextField(required=False, max_length=512)
107 referrer: str = types.TextField(required=False, max_length=512)
108
109 created_at: datetime = types.DateTimeField(create_now=True)
110
111 query: postgres.QuerySet[RedirectLog] = postgres.QuerySet()
112
113 model_options = postgres.Options(
114 ordering=["-created_at"],
115 indexes=[
116 postgres.Index(
117 name="plainredirection_redirectlog_created_at_idx",
118 fields=["created_at"],
119 ),
120 postgres.Index(
121 name="plainredirection_redirectlog_redirect_id_idx",
122 fields=["redirect"],
123 ),
124 ],
125 constraints=[
126 postgres.CheckConstraint(
127 check=postgres.Q(http_status__gte=0),
128 name="plainredirection_redirectlog_http_status_check",
129 ),
130 ],
131 )
132
133 @classmethod
134 def from_redirect(cls, redirect: Redirect, request: Request) -> RedirectLog:
135 from_url = request.build_absolute_uri()
136 to_url = redirect.get_redirect_url(request)
137
138 if not to_url.startswith("http"):
139 to_url = request.build_absolute_uri(to_url)
140
141 if from_url == to_url:
142 raise ValueError("Redirecting to the same URL")
143
144 return cls.query.create(
145 redirect=redirect,
146 from_url=from_url,
147 to_url=to_url,
148 http_status=redirect.http_status,
149 ip_address=request.client_ip,
150 user_agent=request.headers.get("User-Agent", ""),
151 referrer=request.headers.get("Referer", ""),
152 )
153
154
155@postgres.register_model
156class NotFoundLog(postgres.Model):
157 url: str = types.URLField(max_length=512)
158
159 # Request metadata
160 ip_address: str = types.GenericIPAddressField()
161 user_agent: str = types.TextField(required=False, max_length=512)
162 referrer: str = types.TextField(required=False, max_length=512)
163
164 created_at: datetime = types.DateTimeField(create_now=True)
165
166 query: postgres.QuerySet[NotFoundLog] = postgres.QuerySet()
167
168 model_options = postgres.Options(
169 ordering=["-created_at"],
170 indexes=[
171 postgres.Index(
172 name="plainredirection_notfoundlog_created_at_idx",
173 fields=["created_at"],
174 ),
175 ],
176 )
177
178 @classmethod
179 def from_request(cls, request: Request) -> NotFoundLog:
180 return cls.query.create(
181 url=request.build_absolute_uri(),
182 ip_address=request.client_ip,
183 user_agent=request.headers.get("User-Agent", ""),
184 referrer=request.headers.get("Referer", ""),
185 )