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