Coverage for app/backend/src/couchers/utils.py: 93%
206 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import http.cookies
2import re
3import typing
4from collections.abc import Mapping, Sequence
5from datetime import UTC, date, datetime, timedelta, tzinfo
6from email.utils import formatdate
7from typing import TYPE_CHECKING, Any, overload
8from zoneinfo import ZoneInfo
10import regex
11from geoalchemy2 import WKBElement, WKTElement
12from geoalchemy2.shape import from_shape, to_shape
13from google.protobuf.duration_pb2 import Duration
14from google.protobuf.timestamp_pb2 import Timestamp
15from shapely.geometry import Point, Polygon, shape
16from sqlalchemy import Function, cast
17from sqlalchemy.orm import Mapped
18from sqlalchemy.sql import func
19from sqlalchemy.types import DateTime
21from couchers.config import config
22from couchers.constants import (
23 EMAIL_REGEX,
24 PREFERRED_LANGUAGE_COOKIE_EXPIRY,
25 VALID_NAME_CHARACTERS_REGEX,
26 VALID_NAME_MAX_LENGTH,
27 VALID_NAME_MIN_LENGTH,
28 VALID_NAME_NO_SURROUNDING_WHITESPACE_REGEX,
29 VALID_USERNAME_REGEX,
30)
31from couchers.crypto import (
32 create_sofa_id,
33 decode_sofa,
34 decrypt_page_token,
35 encode_sofa,
36 encrypt_page_token,
37)
38from couchers.proto.internal import internal_pb2
40_VALID_NAME_CHARACTERS_PATTERN = regex.compile(VALID_NAME_CHARACTERS_REGEX, regex.UNICODE)
41_VALID_NAME_NO_SURROUNDING_WHITESPACE_PATTERN = regex.compile(VALID_NAME_NO_SURROUNDING_WHITESPACE_REGEX, regex.UNICODE)
43if TYPE_CHECKING:
44 from couchers.models import Geom
47# When a user logs in, they can basically input one of three things: user id, username, or email
48# These are three non-intersecting sets
49# * user_ids are numeric representations in base 10
50# * usernames are alphanumeric + underscores, at least 2 chars long, and don't start with a number,
51# and don't start or end with underscore
52# * emails are just whatever stack overflow says emails are ;)
55def is_valid_user_id(field: str) -> bool:
56 """
57 Checks if it's a string representing a base 10 integer not starting with 0
58 """
59 return re.match(r"[1-9][0-9]*$", field) is not None
62def is_valid_username(field: str) -> bool:
63 """
64 Checks if it's an alphanumeric + underscore, lowercase string, at least
65 two characters long, and starts with a letter, ends with alphanumeric
66 """
67 return re.fullmatch(VALID_USERNAME_REGEX, field) is not None
70def is_valid_name(field: str) -> bool:
71 """
72 Checks that the name satisfies the same rules as the web frontend:
74 * only letters (any Unicode letter), whitespace, apostrophes, and hyphens
75 * no leading or trailing whitespace
76 * 2-100 characters
77 """
78 if len(field) > VALID_NAME_MAX_LENGTH or len(field) < VALID_NAME_MIN_LENGTH:
79 return False
81 return (
82 _VALID_NAME_CHARACTERS_PATTERN.fullmatch(field) is not None
83 and _VALID_NAME_NO_SURROUNDING_WHITESPACE_PATTERN.fullmatch(field) is not None
84 )
87def is_valid_email(field: str) -> bool:
88 return re.match(EMAIL_REGEX, field) is not None
91def Timestamp_from_datetime(dt: datetime) -> Timestamp:
92 if dt.tzinfo is None: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true
93 raise ValueError("Cannot convert a naive datetime to a timestamp.")
95 pb_ts = Timestamp()
96 pb_ts.FromDatetime(dt)
97 return pb_ts
100def Duration_from_timedelta(dt: timedelta) -> Duration:
101 pb_d = Duration()
102 pb_d.FromTimedelta(dt)
103 return pb_d
106def parse_date(date_str: str) -> date | None:
107 """
108 Parses a date-only string in the format "YYYY-MM-DD" returning None if it fails
109 """
110 try:
111 return date.fromisoformat(date_str)
112 except ValueError:
113 return None
116def date_to_api(date_obj: date) -> str:
117 return date_obj.isoformat()
120def to_aware_datetime(ts: Timestamp) -> datetime:
121 """
122 Turns a protobuf Timestamp object into a timezone-aware datetime
123 """
124 return ts.ToDatetime(tzinfo=UTC)
127def to_timezone(value: Timestamp | datetime, timezone: tzinfo) -> datetime:
128 """Returns an instant in time as a datetime in a given timezone."""
129 if isinstance(value, Timestamp):
130 return value.ToDatetime(timezone)
132 if value.tzinfo is None: 132 ↛ 134line 132 didn't jump to line 134 because the condition on line 132 was never true
133 # A naive datetime does not represent a point in time.
134 raise ValueError("Cannot convert a naive datetime to a timezone.")
136 return value.astimezone(timezone)
139def datetime_to_iso8601_local(value: datetime) -> str:
140 """
141 Gets a local ISO 8601 representation of a datetime, without timezone information.
142 This loses information and requires parsers to assume a timezone, so use with care.
143 """
144 return value.replace(tzinfo=None).isoformat()
147def _mockable_now() -> datetime:
148 return datetime.now(tz=UTC)
151def now() -> datetime:
152 # everything that reads the clock goes through this call, so tests can move it in one place
153 # by swapping _mockable_now; see the timewarp fixture
154 return _mockable_now()
157def minimum_allowed_birthdate() -> date:
158 """
159 Most recent birthdate allowed to register (must be 18 years minimum)
161 This approximation works on leap days!
162 """
163 return today() - timedelta(days=365.25 * 18)
166def today() -> date:
167 """
168 Date only in UTC
169 """
170 return now().date()
173def now_in_timezone(tz: str) -> datetime:
174 """
175 tz should be tzdata identifier, e.g. America/New_York
176 """
177 return now().astimezone(ZoneInfo(tz))
180def today_in_timezone(tz: str) -> date:
181 """
182 tz should be tzdata identifier, e.g. America/New_York
183 """
184 return now_in_timezone(tz).date()
187# Note: be very careful with ordering of lat/lng!
188# In a lot of cases they come as (lng, lat), but us humans tend to use them from GPS as (lat, lng)...
189# When entering as EPSG4326, we also need it in (lng, lat)
192def wrap_coordinate(lat: float, lng: float) -> tuple[float, float]:
193 """
194 Wraps (lat, lng) point in the EPSG4326 format
195 """
197 def __wrap_gen(deg: float, ct: float, adj: float) -> float:
198 if deg > ct:
199 deg -= adj
200 if deg < -ct:
201 deg += adj
202 return deg
204 def __wrap_flip(deg: float, ct: float, adj: float) -> float:
205 if deg > ct:
206 deg = -deg + adj
207 if deg < -ct:
208 deg = -deg - adj
209 return deg
211 def __wrap_rem(deg: float, ct: float = 360) -> float:
212 if deg > ct:
213 deg = deg % ct
214 if deg < -ct:
215 deg = deg % -ct
216 return deg
218 if lng < -180 or lng > 180 or lat < -90 or lat > 90:
219 lng = __wrap_rem(lng)
220 lat = __wrap_rem(lat)
221 lng = __wrap_gen(lng, 180, 360)
222 lat = __wrap_flip(lat, 180, 180)
223 lat = __wrap_flip(lat, 90, 180)
224 if lng == -180:
225 lng = 180
226 if lng == -360: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 lng = 0
229 return lat, lng
232def create_coordinate(lat: float, lng: float) -> WKBElement:
233 """
234 Creates a WKT point from a (lat, lng) tuple in EPSG4326 coordinate system (normal GPS-coordinates)
235 """
236 lat, lng = wrap_coordinate(lat, lng)
237 return from_shape(Point(lng, lat), srid=4326)
240def create_polygon_lat_lng(points: list[list[float]]) -> WKBElement:
241 """
242 Creates a EPSG4326 WKT polygon from a list of (lat, lng) tuples
243 """
244 return from_shape(Polygon([(lng, lat) for (lat, lng) in points]), srid=4326)
247def create_polygon_lng_lat(points: list[list[float]]) -> WKBElement:
248 """
249 Creates a EPSG4326 WKT polygon from a list of (lng, lat) tuples
250 """
251 return from_shape(Polygon(points), srid=4326)
254def geojson_to_geom(geojson: dict[str, Any]) -> WKBElement:
255 """
256 Turns GeoJSON to PostGIS geom data in EPSG4326
257 """
258 return from_shape(shape(geojson), srid=4326)
261def to_multi(polygon: WKBElement) -> Function[Any]:
262 return func.ST_Multi(polygon)
265@overload
266def get_coordinates(geom: WKBElement | WKTElement) -> tuple[float, float]: ...
267@overload
268def get_coordinates(geom: None) -> None: ...
271def get_coordinates(geom: WKBElement | WKTElement | None) -> tuple[float, float] | None:
272 """
273 Returns EPSG4326 (lat, lng) pair for a given WKT geom point or None if the input is not truthy
274 """
275 if geom:
276 shp = to_shape(geom)
277 # note the funniness with 4326 normally being (x, y) = (lng, lat)
278 return shp.y, shp.x
279 else:
280 return None
283def http_date(dt: datetime | None = None) -> str:
284 """
285 Format the datetime for HTTP cookies
286 """
287 if not dt:
288 dt = now()
289 return formatdate(dt.timestamp(), usegmt=True)
292def _create_tasty_cookie(name: str, value: Any, expiry: datetime, httponly: bool) -> str:
293 cookie: http.cookies.Morsel[str] = http.cookies.Morsel()
294 cookie.set(name, str(value), str(value))
295 # tell the browser when to stop sending the cookie
296 cookie["expires"] = http_date(expiry)
297 # restrict to our domain, note if there's no domain, it won't include subdomains
298 cookie["domain"] = config.COOKIE_DOMAIN
299 # path so that it's accessible for all API requests, otherwise defaults to something like /org.couchers.auth/
300 cookie["path"] = "/"
301 if config.DEV: 301 ↛ 306line 301 didn't jump to line 306 because the condition on line 301 was always true
302 # send only on requests from first-party domains
303 cookie["samesite"] = "Strict"
304 else:
305 # send on all requests, requires Secure
306 cookie["samesite"] = "None"
307 # only set cookie on HTTPS sites in production
308 cookie["secure"] = True
309 # not accessible from javascript
310 cookie["httponly"] = httponly
312 return cookie.OutputString()
315def create_session_cookies(token: str, user_id: str | int, expiry: datetime) -> list[str]:
316 """
317 Creates our session cookies.
319 We have two: the secure session token (in couchers-sesh) that's inaccessible to javascript, and the user id (in couchers-user-id) which the javascript frontend can access, so that it knows when it's logged in/out
320 """
321 return [
322 _create_tasty_cookie("couchers-sesh", token, expiry, httponly=True),
323 _create_tasty_cookie("couchers-user-id", user_id, expiry, httponly=False),
324 ]
327def create_lang_cookie(lang: str) -> list[str]:
328 return [
329 _create_tasty_cookie("NEXT_LOCALE", lang, expiry=(now() + PREFERRED_LANGUAGE_COOKIE_EXPIRY), httponly=False)
330 ]
333def _parse_cookie(headers: Mapping[str, str | bytes], cookie_name: str) -> str | None:
334 """
335 Helper to parse a cookie value from headers by name, returning None if not found.
336 """
337 if "cookie" not in headers:
338 return None
340 cookie_str = typing.cast(str, headers["cookie"])
341 cookie = http.cookies.SimpleCookie(cookie_str).get(cookie_name)
343 if not cookie:
344 return None
346 return cookie.value
349def parse_session_cookie(headers: Mapping[str, str | bytes]) -> str | None:
350 """
351 Returns our session cookie value (aka token) or None
352 """
353 return _parse_cookie(headers, "couchers-sesh")
356def parse_user_id_cookie(headers: Mapping[str, str | bytes]) -> str | None:
357 """
358 Returns our user id cookie value or None
359 """
360 return _parse_cookie(headers, "couchers-user-id")
363def parse_ui_lang_cookie(headers: Mapping[str, str | bytes]) -> str | None:
364 """
365 Returns language cookie or None
366 """
367 return _parse_cookie(headers, "NEXT_LOCALE")
370def parse_api_key(headers: Mapping[str, str | bytes]) -> str | None:
371 """
372 Returns a bearer token (API key) from the `authorization` header, or None if invalid/not present
373 """
374 if "authorization" not in headers: 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true
375 return None
377 authorization = headers["authorization"]
378 if isinstance(authorization, bytes): 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true
379 authorization = authorization.decode("utf-8")
381 if not authorization.startswith("Bearer "):
382 return None
384 return authorization[7:]
387def parse_sofa_cookie(headers: Mapping[str, str | bytes]) -> str | None:
388 cookie_value = _parse_cookie(headers, "sofa")
389 if not cookie_value:
390 return None
392 try:
393 decode_sofa(cookie_value)
394 return cookie_value
395 except Exception:
396 return None
399def generate_sofa_cookie() -> tuple[str, str]:
400 sofa_value = encode_sofa(
401 create_sofa_id(),
402 internal_pb2.SofaPayload(
403 version=1,
404 created=Timestamp_from_datetime(now()),
405 ),
406 )
407 return sofa_value, _create_tasty_cookie("sofa", sofa_value, now() + timedelta(days=10000), httponly=True)
410def remove_duplicates_retain_order[T](list_: Sequence[T]) -> list[T]:
411 out = []
412 for item in list_:
413 if item not in out:
414 out.append(item)
415 return out
418def date_in_timezone(date_: Mapped[date | None], timezone: str) -> Function[Any]:
419 """
420 Given a naive postgres date object (postgres doesn't have tzd dates), returns a timezone-aware timestamp for the
421 start of that date in that timezone. E.g., if postgres is in 'America/New_York',
423 SET SESSION TIME ZONE 'America/New_York';
425 CREATE TABLE tz_trouble (to_date date, timezone text);
427 INSERT INTO tz_trouble(to_date, timezone) VALUES
428 ('2021-03-10'::date, 'Australia/Sydney'),
429 ('2021-03-20'::date, 'Europe/Berlin'),
430 ('2021-04-15'::date, 'America/New_York');
432 SELECT timezone(timezone, to_date::timestamp) FROM tz_trouble;
434 The result is:
436 timezone
437 ------------------------
438 2021-03-09 08:00:00-05
439 2021-03-19 19:00:00-04
440 2021-04-15 00:00:00-04
441 """
442 return func.timezone(timezone, cast(date_, DateTime(timezone=False)))
445def millis_from_dt(dt: datetime) -> int:
446 return round(1000 * dt.timestamp())
449def dt_from_millis(millis: int) -> datetime:
450 return datetime.fromtimestamp(millis / 1000, tz=UTC)
453def dt_to_page_token(dt: datetime) -> str:
454 """
455 Python has datetime resolution equal to 1 micro, as does postgres
457 We pray to deities that this never changes
458 """
459 assert datetime.resolution == timedelta(microseconds=1)
460 return encrypt_page_token(str(round(1_000_000 * dt.timestamp())))
463def dt_from_page_token(page_token: str) -> datetime:
464 # see above comment
465 return datetime.fromtimestamp(int(decrypt_page_token(page_token)) / 1_000_000, tz=UTC)
468def dt_id_to_page_token(dt: datetime, id_: int) -> str:
469 # see above comment about resolution
470 return encrypt_page_token(f"{round(1_000_000 * dt.timestamp())}:{id_}")
473def dt_id_from_page_token(page_token: str) -> tuple[datetime, int]:
474 micros, id_ = decrypt_page_token(page_token).split(":")
475 return datetime.fromtimestamp(int(micros) / 1_000_000, tz=UTC), int(id_)
478def last_active_coarsen(dt: datetime) -> datetime:
479 """
480 Coarsens a "last active" time to the accuracy we use for last active times, currently to the last hour, e.g. if the current time is 27th June 2021, 16:53 UTC, this returns 27th June 2021, 16:00 UTC
481 """
482 return dt.replace(minute=0, second=0, microsecond=0)
485def not_none[T](x: T | None) -> T:
486 if x is None: 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true
487 raise ValueError("Expected a value but got None")
488 return x
491def is_geom(x: Geom | None) -> Geom:
492 """not_none does not work with unions."""
493 if x is None: 493 ↛ 494line 493 didn't jump to line 494 because the condition on line 493 was never true
494 raise ValueError("Expected a Geom but got None")
495 return x