Coverage for app/backend/src/couchers/utils.py: 94%
211 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +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 cookie["samesite"] = config.COOKIE_SAMESITE
302 if not config.DEV:
303 # only set cookie on HTTPS sites in production
304 cookie["secure"] = True
305 cookie["partitioned"] = config.COOKIE_PARTITIONED
306 # not accessible from javascript
307 cookie["httponly"] = httponly
309 return cookie.OutputString()
312def create_session_cookies(token: str, user_id: str | int, expiry: datetime) -> list[str]:
313 """
314 Creates our session cookies.
316 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
317 """
318 return [
319 _create_tasty_cookie("couchers-sesh", token, expiry, httponly=True),
320 _create_tasty_cookie("couchers-user-id", user_id, expiry, httponly=False),
321 ]
324def create_lang_cookie(lang: str) -> list[str]:
325 return [
326 _create_tasty_cookie("NEXT_LOCALE", lang, expiry=(now() + PREFERRED_LANGUAGE_COOKIE_EXPIRY), httponly=False)
327 ]
330def _parse_cookie(headers: Mapping[str, str | bytes], cookie_name: str) -> str | None:
331 """
332 Helper to parse a cookie value from headers by name, returning None if not found.
333 """
334 if "cookie" not in headers:
335 return None
337 cookie_str = typing.cast(str, headers["cookie"])
338 cookie = http.cookies.SimpleCookie(cookie_str).get(cookie_name)
340 if not cookie:
341 return None
343 return cookie.value
346def parse_session_cookie(headers: Mapping[str, str | bytes]) -> str | None:
347 """
348 Returns our session cookie value (aka token) or None
349 """
350 return _parse_cookie(headers, "couchers-sesh")
353def parse_user_id_cookie(headers: Mapping[str, str | bytes]) -> str | None:
354 """
355 Returns our user id cookie value or None
356 """
357 return _parse_cookie(headers, "couchers-user-id")
360def parse_ui_lang_cookie(headers: Mapping[str, str | bytes]) -> str | None:
361 """
362 Returns language cookie or None
363 """
364 return _parse_cookie(headers, "NEXT_LOCALE")
367def parse_api_key(headers: Mapping[str, str | bytes]) -> str | None:
368 """
369 Returns a bearer token (API key) from the `authorization` header, or None if invalid/not present
370 """
371 if "authorization" not in headers: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 return None
374 authorization = headers["authorization"]
375 if isinstance(authorization, bytes): 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 authorization = authorization.decode("utf-8")
378 if not authorization.startswith("Bearer "):
379 return None
381 return authorization[7:]
384def parse_sofa_cookie(headers: Mapping[str, str | bytes]) -> str | None:
385 cookie_value = _parse_cookie(headers, "sofa")
386 if not cookie_value:
387 return None
389 try:
390 decode_sofa(cookie_value)
391 return cookie_value
392 except Exception:
393 return None
396def generate_sofa_cookie() -> tuple[str, str]:
397 sofa_value = encode_sofa(
398 create_sofa_id(),
399 internal_pb2.SofaPayload(
400 version=1,
401 created=Timestamp_from_datetime(now()),
402 ),
403 )
404 return sofa_value, _create_tasty_cookie("sofa", sofa_value, now() + timedelta(days=10000), httponly=True)
407def remove_duplicates_retain_order[T](list_: Sequence[T]) -> list[T]:
408 out = []
409 for item in list_:
410 if item not in out:
411 out.append(item)
412 return out
415def date_in_timezone(date_: Mapped[date | None], timezone: str | Mapped[str]) -> Function[Any]:
416 """
417 Given a naive postgres date object (postgres doesn't have tzd dates), returns a timezone-aware timestamp for the
418 start of that date in that timezone. E.g., if postgres is in 'America/New_York',
420 SET SESSION TIME ZONE 'America/New_York';
422 CREATE TABLE tz_trouble (to_date date, timezone text);
424 INSERT INTO tz_trouble(to_date, timezone) VALUES
425 ('2021-03-10'::date, 'Australia/Sydney'),
426 ('2021-03-20'::date, 'Europe/Berlin'),
427 ('2021-04-15'::date, 'America/New_York');
429 SELECT timezone(timezone, to_date::timestamp) FROM tz_trouble;
431 The result is:
433 timezone
434 ------------------------
435 2021-03-09 08:00:00-05
436 2021-03-19 19:00:00-04
437 2021-04-15 00:00:00-04
438 """
439 return func.timezone(timezone, cast(date_, DateTime(timezone=False)))
442def millis_from_dt(dt: datetime) -> int:
443 return round(1000 * dt.timestamp())
446def dt_from_millis(millis: int) -> datetime:
447 return datetime.fromtimestamp(millis / 1000, tz=UTC)
450def dt_to_page_token(dt: datetime) -> str:
451 """
452 Python has datetime resolution equal to 1 micro, as does postgres
454 We pray to deities that this never changes
455 """
456 assert datetime.resolution == timedelta(microseconds=1)
457 return encrypt_page_token(str(round(1_000_000 * dt.timestamp())))
460def dt_from_page_token(page_token: str) -> datetime:
461 # see above comment
462 return datetime.fromtimestamp(int(decrypt_page_token(page_token)) / 1_000_000, tz=UTC)
465def dt_id_to_page_token(dt: datetime, id_: int) -> str:
466 # see above comment about resolution
467 return encrypt_page_token(f"{round(1_000_000 * dt.timestamp())}:{id_}")
470def dt_id_from_page_token(page_token: str) -> tuple[datetime, int]:
471 micros, id_ = decrypt_page_token(page_token).split(":")
472 return datetime.fromtimestamp(int(micros) / 1_000_000, tz=UTC), int(id_)
475def date_id_to_page_token(date_: date, id_: int) -> str:
476 return encrypt_page_token(f"{date_.isoformat()}:{id_}")
479def date_id_from_page_token(page_token: str) -> tuple[date, int]:
480 date_str, id_ = decrypt_page_token(page_token).split(":")
481 return date.fromisoformat(date_str), int(id_)
484def last_active_coarsen(dt: datetime) -> datetime:
485 """
486 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
487 """
488 return dt.replace(minute=0, second=0, microsecond=0)
491def not_none[T](x: T | None) -> T:
492 if x is None: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true
493 raise ValueError("Expected a value but got None")
494 return x
497def is_geom(x: Geom | None) -> Geom:
498 """not_none does not work with unions."""
499 if x is None: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 raise ValueError("Expected a Geom but got None")
501 return x