Browse Source

Type annotations: use standard generics (PEP 585)

As of Python 3.9, it's no longer necessary to import things like List
and Dict from the typing module, and we can just use the built-in types
like this.
merge-requests/135/head
Deimos 3 years ago
parent
commit
ed38ce5790
  1. 4
      tildes/consumers/post_processing_script_runner.py
  2. 3
      tildes/consumers/site_icon_downloader.py
  3. 2
      tildes/consumers/topic_embedly_extractor.py
  4. 7
      tildes/consumers/topic_metadata_generator.py
  5. 2
      tildes/consumers/topic_youtube_scraper.py
  6. 4
      tildes/tildes/__init__.py
  7. 3
      tildes/tildes/auth.py
  8. 4
      tildes/tildes/database.py
  9. 4
      tildes/tildes/enums.py
  10. 4
      tildes/tildes/lib/auth.py
  11. 11
      tildes/tildes/lib/database.py
  12. 9
      tildes/tildes/lib/event_stream.py
  13. 3
      tildes/tildes/lib/lua.py
  14. 30
      tildes/tildes/lib/markdown.py
  15. 7
      tildes/tildes/lib/ratelimit.py
  16. 4
      tildes/tildes/lib/site_info.py
  17. 7
      tildes/tildes/lib/string.py
  18. 2
      tildes/tildes/metrics.py
  19. 3
      tildes/tildes/models/comment/comment.py
  20. 7
      tildes/tildes/models/comment/comment_notification.py
  21. 13
      tildes/tildes/models/comment/comment_tree.py
  22. 6
      tildes/tildes/models/database_model.py
  23. 8
      tildes/tildes/models/group/group.py
  24. 4
      tildes/tildes/models/group/group_wiki_page.py
  25. 8
      tildes/tildes/models/log/log.py
  26. 5
      tildes/tildes/models/message/message.py
  27. 3
      tildes/tildes/models/model_query.py
  28. 9
      tildes/tildes/models/pagination.py
  29. 15
      tildes/tildes/models/topic/topic.py
  30. 3
      tildes/tildes/models/topic/topic_query.py
  31. 6
      tildes/tildes/models/topic/topic_schedule.py
  32. 8
      tildes/tildes/models/user/user.py
  33. 10
      tildes/tildes/request_methods.py
  34. 7
      tildes/tildes/schemas/fields.py
  35. 5
      tildes/tildes/schemas/topic.py
  36. 6
      tildes/tildes/scrapers/embedly_scraper.py
  37. 6
      tildes/tildes/scrapers/youtube_scraper.py
  38. 2
      tildes/tildes/tweens.py
  39. 6
      tildes/tildes/typing.py
  40. 2
      tildes/tildes/views/api/web/exceptions.py
  41. 4
      tildes/tildes/views/bookmarks.py
  42. 5
      tildes/tildes/views/decorators.py
  43. 2
      tildes/tildes/views/exceptions.py
  44. 6
      tildes/tildes/views/financials.py
  45. 12
      tildes/tildes/views/user.py
  46. 4
      tildes/tildes/views/votes.py

4
tildes/consumers/post_processing_script_runner.py

@ -3,7 +3,7 @@
"""Consumer that runs processing scripts on posts.""" """Consumer that runs processing scripts on posts."""
from typing import Type, Union
from typing import Union
from sqlalchemy import desc from sqlalchemy import desc
from sqlalchemy.sql.expression import or_ from sqlalchemy.sql.expression import or_
@ -23,7 +23,7 @@ class PostProcessingScriptRunner(EventStreamConsumer):
def process_message(self, message: Message) -> None: def process_message(self, message: Message) -> None:
"""Process a message from the stream.""" """Process a message from the stream."""
wrapper_class: Union[Type[TopicScriptingWrapper], Type[CommentScriptingWrapper]]
wrapper_class: Union[type[TopicScriptingWrapper], type[CommentScriptingWrapper]]
if "topic_id" in message.fields: if "topic_id" in message.fields:
post = ( post = (

3
tildes/consumers/site_icon_downloader.py

@ -3,9 +3,10 @@
"""Consumer that downloads site icons using Embedly scraper data.""" """Consumer that downloads site icons using Embedly scraper data."""
from collections.abc import Sequence
from io import BytesIO from io import BytesIO
from os import path from os import path
from typing import Optional, Sequence
from typing import Optional
import publicsuffix import publicsuffix
import requests import requests

2
tildes/consumers/topic_embedly_extractor.py

@ -4,8 +4,8 @@
"""Consumer that fetches data from Embedly's Extract API for link topics.""" """Consumer that fetches data from Embedly's Extract API for link topics."""
import os import os
from collections.abc import Sequence
from datetime import timedelta from datetime import timedelta
from typing import Sequence
from pyramid.paster import get_appsettings from pyramid.paster import get_appsettings
from requests.exceptions import HTTPError, Timeout from requests.exceptions import HTTPError, Timeout

7
tildes/consumers/topic_metadata_generator.py

@ -3,7 +3,8 @@
"""Consumer that generates content_metadata for topics.""" """Consumer that generates content_metadata for topics."""
from typing import Any, Dict, Sequence
from collections.abc import Sequence
from typing import Any
from ipaddress import ip_address from ipaddress import ip_address
import publicsuffix import publicsuffix
@ -61,7 +62,7 @@ class TopicMetadataGenerator(EventStreamConsumer):
) )
@staticmethod @staticmethod
def _generate_text_metadata(topic: Topic) -> Dict[str, Any]:
def _generate_text_metadata(topic: Topic) -> dict[str, Any]:
"""Generate metadata for a text topic (word count and excerpt).""" """Generate metadata for a text topic (word count and excerpt)."""
if not topic.rendered_html: if not topic.rendered_html:
return {} return {}
@ -81,7 +82,7 @@ class TopicMetadataGenerator(EventStreamConsumer):
except ValueError: except ValueError:
return False return False
def _generate_link_metadata(self, topic: Topic) -> Dict[str, Any]:
def _generate_link_metadata(self, topic: Topic) -> dict[str, Any]:
"""Generate metadata for a link topic (domain).""" """Generate metadata for a link topic (domain)."""
if not topic.link: if not topic.link:
return {} return {}

2
tildes/consumers/topic_youtube_scraper.py

@ -4,8 +4,8 @@
"""Consumer that fetches data from YouTube's data API for relevant link topics.""" """Consumer that fetches data from YouTube's data API for relevant link topics."""
import os import os
from collections.abc import Sequence
from datetime import timedelta from datetime import timedelta
from typing import Sequence
from pyramid.paster import get_appsettings from pyramid.paster import get_appsettings
from requests.exceptions import HTTPError, Timeout from requests.exceptions import HTTPError, Timeout

4
tildes/tildes/__init__.py

@ -3,8 +3,6 @@
"""Configure and initialize the Pyramid app.""" """Configure and initialize the Pyramid app."""
from typing import Dict
import sentry_sdk import sentry_sdk
from marshmallow.exceptions import ValidationError from marshmallow.exceptions import ValidationError
from paste.deploy.config import PrefixMiddleware from paste.deploy.config import PrefixMiddleware
@ -13,7 +11,7 @@ from sentry_sdk.integrations.pyramid import PyramidIntegration
from webassets import Bundle from webassets import Bundle
def main(global_config: Dict[str, str], **settings: str) -> PrefixMiddleware:
def main(global_config: dict[str, str], **settings: str) -> PrefixMiddleware:
"""Configure and return a Pyramid WSGI application.""" """Configure and return a Pyramid WSGI application."""
config = Configurator(settings=settings) config = Configurator(settings=settings)

3
tildes/tildes/auth.py

@ -3,7 +3,8 @@
"""Configuration and functionality related to authentication/authorization.""" """Configuration and functionality related to authentication/authorization."""
from typing import Any, Optional, Sequence
from collections.abc import Sequence
from typing import Any, Optional
from pyramid.authentication import SessionAuthenticationPolicy from pyramid.authentication import SessionAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy from pyramid.authorization import ACLAuthorizationPolicy

4
tildes/tildes/database.py

@ -3,7 +3,7 @@
"""Contains the database-related config updates and request methods.""" """Contains the database-related config updates and request methods."""
from typing import Callable, Type
from collections.abc import Callable
from pyramid.config import Configurator from pyramid.config import Configurator
from pyramid.request import Request from pyramid.request import Request
@ -31,7 +31,7 @@ def obtain_lock(request: Request, lock_space: str, lock_value: int) -> None:
obtain_transaction_lock(request.db_session, lock_space, lock_value) obtain_transaction_lock(request.db_session, lock_space, lock_value)
def query_factory(request: Request, model_cls: Type[DatabaseModel]) -> ModelQuery:
def query_factory(request: Request, model_cls: type[DatabaseModel]) -> ModelQuery:
"""Return a ModelQuery or subclass depending on model_cls specified.""" """Return a ModelQuery or subclass depending on model_cls specified."""
if model_cls == Comment: if model_cls == Comment:
return CommentQuery(request) return CommentQuery(request)

4
tildes/tildes/enums.py

@ -5,7 +5,7 @@
import enum import enum
from datetime import timedelta from datetime import timedelta
from typing import Any, List, Optional
from typing import Any, Optional
from tildes.lib.datetime import utc_from_timestamp from tildes.lib.datetime import utc_from_timestamp
@ -109,7 +109,7 @@ class ContentMetadataFields(enum.Enum):
def detail_fields_for_content_type( def detail_fields_for_content_type(
cls, cls,
content_type: "TopicContentType", content_type: "TopicContentType",
) -> List["ContentMetadataFields"]:
) -> list["ContentMetadataFields"]:
"""Return a list of fields to display for detail about a particular type.""" """Return a list of fields to display for detail about a particular type."""
if content_type is TopicContentType.ARTICLE: if content_type is TopicContentType.ARTICLE:
return [cls.WORD_COUNT, cls.PUBLISHED] return [cls.WORD_COUNT, cls.PUBLISHED]

4
tildes/tildes/lib/auth.py

@ -3,7 +3,7 @@
"""Functions to help with authorization, such as generating ACLs.""" """Functions to help with authorization, such as generating ACLs."""
from typing import List, Optional
from typing import Optional
from pyramid.security import Allow, Deny from pyramid.security import Allow, Deny
@ -14,7 +14,7 @@ def aces_for_permission(
required_permission: str, required_permission: str,
group_id: Optional[int] = None, group_id: Optional[int] = None,
granted_permission: Optional[str] = None, granted_permission: Optional[str] = None,
) -> List[AceType]:
) -> list[AceType]:
"""Return the ACEs for manually-granted (or denied) entries in UserPermissions.""" """Return the ACEs for manually-granted (or denied) entries in UserPermissions."""
aces = [] aces = []

11
tildes/tildes/lib/database.py

@ -4,7 +4,8 @@
"""Constants/classes/functions related to the database.""" """Constants/classes/functions related to the database."""
import enum import enum
from typing import Any, Callable, List, Optional
from collections.abc import Callable
from typing import Any, Optional
from dateutil.rrule import rrule, rrulestr from dateutil.rrule import rrule, rrulestr
from pyramid.paster import bootstrap from pyramid.paster import bootstrap
@ -106,7 +107,7 @@ class ArrayOfLtree(ARRAY):
"""Return a conversion function for processing result row values.""" """Return a conversion function for processing result row values."""
super_rp = super().result_processor(dialect, coltype) super_rp = super().result_processor(dialect, coltype)
def handle_raw_string(value: str) -> List[str]:
def handle_raw_string(value: str) -> list[str]:
if not (value.startswith("{") and value.endswith("}")): if not (value.startswith("{") and value.endswith("}")):
raise ValueError("%s is not an array value" % value) raise ValueError("%s is not an array value" % value)
@ -119,7 +120,7 @@ class ArrayOfLtree(ARRAY):
return value.split(",") return value.split(",")
def process(value: Optional[str]) -> Optional[List[str]]:
def process(value: Optional[str]) -> Optional[list[str]]:
if value is None: if value is None:
return None return None
@ -183,10 +184,10 @@ class TagList(TypeDecorator):
impl = ArrayOfLtree impl = ArrayOfLtree
def process_bind_param(self, value: str, dialect: Dialect) -> List[Ltree]:
def process_bind_param(self, value: str, dialect: Dialect) -> list[Ltree]:
"""Convert the value to ltree[] for storing.""" """Convert the value to ltree[] for storing."""
return [Ltree(tag.replace(" ", "_")) for tag in value] return [Ltree(tag.replace(" ", "_")) for tag in value]
def process_result_value(self, value: List[Ltree], dialect: Dialect) -> List[str]:
def process_result_value(self, value: list[Ltree], dialect: Dialect) -> list[str]:
"""Convert the stored value to a list of strings.""" """Convert the stored value to a list of strings."""
return [str(tag).replace("_", " ") for tag in value] return [str(tag).replace("_", " ") for tag in value]

9
tildes/tildes/lib/event_stream.py

@ -5,8 +5,9 @@
import os import os
from abc import abstractmethod from abc import abstractmethod
from collections.abc import Sequence
from configparser import ConfigParser from configparser import ConfigParser
from typing import Any, Dict, List, Optional, Sequence
from typing import Any, Optional
from prometheus_client import CollectorRegistry, Counter, start_http_server from prometheus_client import CollectorRegistry, Counter, start_http_server
from redis import Redis, ResponseError from redis import Redis, ResponseError
@ -22,7 +23,7 @@ class Message:
"""Represents a single message taken from a stream.""" """Represents a single message taken from a stream."""
def __init__( def __init__(
self, redis: Redis, stream: str, message_id: str, fields: Dict[str, str]
self, redis: Redis, stream: str, message_id: str, fields: dict[str, str]
): ):
"""Initialize a new message from a Redis stream.""" """Initialize a new message from a Redis stream."""
self.redis = redis self.redis = redis
@ -181,7 +182,7 @@ class EventStreamConsumer:
message_ids=[entry["message_id"]], message_ids=[entry["message_id"]],
) )
def _xreadgroup_response_to_messages(self, response: Any) -> List[Message]:
def _xreadgroup_response_to_messages(self, response: Any) -> list[Message]:
"""Convert a response from XREADGROUP to a list of Messages.""" """Convert a response from XREADGROUP to a list of Messages."""
messages = [] messages = []
@ -204,7 +205,7 @@ class EventStreamConsumer:
return messages return messages
def _get_messages(self, pending: bool = False) -> List[Message]:
def _get_messages(self, pending: bool = False) -> list[Message]:
"""Get any messages from the streams for this consumer. """Get any messages from the streams for this consumer.
This method will return at most one message from each of the source streams per This method will return at most one message from each of the source streams per

3
tildes/tildes/lib/lua.py

@ -3,8 +3,9 @@
"""Functions and classes related to Lua scripting.""" """Functions and classes related to Lua scripting."""
from collections.abc import Callable
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Optional
from typing import Any, Optional
from lupa import LuaError, LuaRuntime from lupa import LuaError, LuaRuntime

30
tildes/tildes/lib/markdown.py

@ -4,19 +4,9 @@
"""Functions/constants related to markdown handling.""" """Functions/constants related to markdown handling."""
import re import re
from collections.abc import Callable, Iterator
from functools import partial from functools import partial
from typing import (
Any,
Callable,
Dict,
Iterator,
List,
Match,
Optional,
Pattern,
Tuple,
Union,
)
from typing import Any, Optional, Union
import bleach import bleach
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
@ -105,7 +95,7 @@ ALLOWED_HTML_TAGS = (
) )
ALLOWED_LINK_PROTOCOLS = ("gemini", "http", "https", "mailto") ALLOWED_LINK_PROTOCOLS = ("gemini", "http", "https", "mailto")
ALLOWED_HTML_ATTRIBUTES_DEFAULT: Dict[str, Union[List[str], Callable]] = {
ALLOWED_HTML_ATTRIBUTES_DEFAULT: dict[str, Union[list[str], Callable]] = {
"a": ["href", "title"], "a": ["href", "title"],
"details": ["open"], "details": ["open"],
"ol": ["start"], "ol": ["start"],
@ -234,7 +224,7 @@ class CodeHtmlFormatter(HtmlFormatter):
<code class="highlight">...</code> instead (assumes a <pre> is already present). <code class="highlight">...</code> instead (assumes a <pre> is already present).
""" """
def wrap(self, source: Any, outfile: Any) -> Iterator[Tuple[int, str]]:
def wrap(self, source: Any, outfile: Any) -> Iterator[tuple[int, str]]:
"""Wrap the highlighted tokens with the <code> tag.""" """Wrap the highlighted tokens with the <code> tag."""
# pylint: disable=unused-argument # pylint: disable=unused-argument
yield (0, '<code class="highlight">') yield (0, '<code class="highlight">')
@ -311,7 +301,7 @@ class LinkifyFilter(Filter):
SUBREDDIT_REFERENCE_REGEX = re.compile(r"(?<!\w)/?r/(\w+)\b") SUBREDDIT_REFERENCE_REGEX = re.compile(r"(?<!\w)/?r/(\w+)\b")
def __init__( def __init__(
self, source: NonRecursiveTreeWalker, skip_tags: Optional[List[str]] = None
self, source: NonRecursiveTreeWalker, skip_tags: Optional[list[str]] = None
): ):
"""Initialize a linkification filter to apply to HTML. """Initialize a linkification filter to apply to HTML.
@ -385,8 +375,8 @@ class LinkifyFilter(Filter):
@staticmethod @staticmethod
def _linkify_tokens( def _linkify_tokens(
tokens: List[dict], filter_regex: Pattern, linkify_function: Callable
) -> List[dict]:
tokens: list[dict], filter_regex: re.Pattern, linkify_function: Callable
) -> list[dict]:
"""Check tokens for text that matches a regex and linkify it. """Check tokens for text that matches a regex and linkify it.
The `filter_regex` argument should be a compiled pattern that will be applied to The `filter_regex` argument should be a compiled pattern that will be applied to
@ -434,7 +424,7 @@ class LinkifyFilter(Filter):
return new_tokens return new_tokens
@staticmethod @staticmethod
def _tokenize_group_match(match: Match) -> List[dict]:
def _tokenize_group_match(match: re.Match) -> list[dict]:
"""Convert a potential group reference into HTML tokens.""" """Convert a potential group reference into HTML tokens."""
# convert the potential group path to lowercase to allow people to use incorrect # convert the potential group path to lowercase to allow people to use incorrect
# casing but still have it link properly # casing but still have it link properly
@ -465,7 +455,7 @@ class LinkifyFilter(Filter):
return [{"type": "Characters", "data": match[0]}] return [{"type": "Characters", "data": match[0]}]
@staticmethod @staticmethod
def _tokenize_username_match(match: Match) -> List[dict]:
def _tokenize_username_match(match: re.Match) -> list[dict]:
"""Convert a potential username reference into HTML tokens.""" """Convert a potential username reference into HTML tokens."""
# if it's a valid username, convert to <a> # if it's a valid username, convert to <a>
if is_valid_username(match[1]): if is_valid_username(match[1]):
@ -486,7 +476,7 @@ class LinkifyFilter(Filter):
return [{"type": "Characters", "data": match[0]}] return [{"type": "Characters", "data": match[0]}]
@staticmethod @staticmethod
def _tokenize_subreddit_match(match: Match) -> List[dict]:
def _tokenize_subreddit_match(match: re.Match) -> list[dict]:
"""Convert a subreddit reference into HTML tokens.""" """Convert a subreddit reference into HTML tokens."""
return [ return [
{ {

7
tildes/tildes/lib/ratelimit.py

@ -4,9 +4,10 @@
"""Classes and constants related to rate-limited actions.""" """Classes and constants related to rate-limited actions."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence
from datetime import timedelta from datetime import timedelta
from ipaddress import ip_address from ipaddress import ip_address
from typing import Any, List, Optional, Sequence
from typing import Any, Optional
from pyramid.response import Response from pyramid.response import Response
from redis import Redis from redis import Redis
@ -69,7 +70,7 @@ class RateLimitResult:
) )
@classmethod @classmethod
def from_redis_cell_result(cls, result: List[int]) -> RateLimitResult:
def from_redis_cell_result(cls, result: list[int]) -> RateLimitResult:
"""Convert the response from CL.THROTTLE command to a RateLimitResult. """Convert the response from CL.THROTTLE command to a RateLimitResult.
CL.THROTTLE responds with an array of 5 integers: CL.THROTTLE responds with an array of 5 integers:
@ -228,7 +229,7 @@ class RateLimitedAction:
return ":".join(parts) return ":".join(parts)
def _call_redis_command(self, key: str) -> List[int]:
def _call_redis_command(self, key: str) -> list[int]:
"""Call the redis-cell CL.THROTTLE command for this action.""" """Call the redis-cell CL.THROTTLE command for this action."""
return self.redis.execute_command( return self.redis.execute_command(
"CL.THROTTLE", "CL.THROTTLE",

4
tildes/tildes/lib/site_info.py

@ -3,7 +3,7 @@
"""Library code related to displaying info about individual websites.""" """Library code related to displaying info about individual websites."""
from typing import List, Optional
from typing import Optional
from tildes.enums import ContentMetadataFields, TopicContentType from tildes.enums import ContentMetadataFields, TopicContentType
@ -22,7 +22,7 @@ class SiteInfo:
self.show_author = show_author self.show_author = show_author
self.content_type = content_type self.content_type = content_type
def content_source(self, authors: Optional[List[str]] = None) -> str:
def content_source(self, authors: Optional[list[str]] = None) -> str:
"""Return a string representing the "source" of content on this site. """Return a string representing the "source" of content on this site.
If the site isn't one that needs to show its author, this is just its name. If the site isn't one that needs to show its author, this is just its name.

7
tildes/tildes/lib/string.py

@ -5,7 +5,8 @@
import re import re
import unicodedata import unicodedata
from typing import Iterator, List, Optional
from collections.abc import Iterator
from typing import Optional
from urllib.parse import quote from urllib.parse import quote
from xml.etree.ElementTree import Element from xml.etree.ElementTree import Element
@ -225,10 +226,10 @@ def separate_string(original: str, separator: str, segment_size: int) -> str:
return separated return separated
def extract_text_from_html(html: str, skip_tags: Optional[List[str]] = None) -> str:
def extract_text_from_html(html: str, skip_tags: Optional[list[str]] = None) -> str:
"""Extract plain text content from the elements inside an HTML string.""" """Extract plain text content from the elements inside an HTML string."""
def extract_text(element: Element, skip_tags: List[str]) -> Iterator[str]:
def extract_text(element: Element, skip_tags: list[str]) -> Iterator[str]:
"""Extract text recursively from elements, optionally skipping some tags. """Extract text recursively from elements, optionally skipping some tags.
This function is Python's xml.etree.ElementTree.Element.itertext() but with the This function is Python's xml.etree.ElementTree.Element.itertext() but with the

2
tildes/tildes/metrics.py

@ -7,7 +7,7 @@
# checks to avoid errors # checks to avoid errors
# pylint: disable=no-value-for-parameter,redundant-keyword-arg # pylint: disable=no-value-for-parameter,redundant-keyword-arg
from typing import Callable
from collections.abc import Callable
from prometheus_client import Counter, Histogram, Summary from prometheus_client import Counter, Histogram, Summary

3
tildes/tildes/models/comment/comment.py

@ -4,8 +4,9 @@
"""Contains the Comment class.""" """Contains the Comment class."""
from collections import Counter from collections import Counter
from collections.abc import Sequence
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Optional, Sequence, TYPE_CHECKING, Union
from typing import Any, Optional, TYPE_CHECKING, Union
from pyramid.security import ( from pyramid.security import (
Allow, Allow,

7
tildes/tildes/models/comment/comment_notification.py

@ -5,7 +5,6 @@
import re import re
from datetime import datetime from datetime import datetime
from typing import List, Tuple
from pyramid.security import Allow, DENY_ALL from pyramid.security import Allow, DENY_ALL
from sqlalchemy import BigInteger, Boolean, Column, ForeignKey, TIMESTAMP from sqlalchemy import BigInteger, Boolean, Column, ForeignKey, TIMESTAMP
@ -120,7 +119,7 @@ class CommentNotification(DatabaseModel):
@classmethod @classmethod
def get_mentions_for_comment( def get_mentions_for_comment(
cls, db_session: Session, comment: Comment cls, db_session: Session, comment: Comment
) -> List["CommentNotification"]:
) -> list["CommentNotification"]:
"""Get a list of notifications for user mentions in the comment.""" """Get a list of notifications for user mentions in the comment."""
notifications = [] notifications = []
@ -160,8 +159,8 @@ class CommentNotification(DatabaseModel):
def prevent_duplicate_notifications( def prevent_duplicate_notifications(
db_session: Session, db_session: Session,
comment: Comment, comment: Comment,
new_notifications: List["CommentNotification"],
) -> Tuple[List["CommentNotification"], List["CommentNotification"]]:
new_notifications: list["CommentNotification"],
) -> tuple[list["CommentNotification"], list["CommentNotification"]]:
"""Filter new notifications for edited comments. """Filter new notifications for edited comments.
Protect against sending a notification for the same comment to the same user Protect against sending a notification for the same comment to the same user

13
tildes/tildes/models/comment/comment_tree.py

@ -4,8 +4,9 @@
"""Contains the CommentTree and CommentInTree classes.""" """Contains the CommentTree and CommentInTree classes."""
from collections import Counter from collections import Counter
from collections.abc import Iterator, Sequence
from datetime import datetime from datetime import datetime
from typing import Iterator, List, Optional, Sequence, Tuple
from typing import Optional
from prometheus_client import Histogram from prometheus_client import Histogram
from wrapt import ObjectProxy from wrapt import ObjectProxy
@ -27,7 +28,7 @@ class CommentTree:
viewer: Optional[User] = None, viewer: Optional[User] = None,
): ):
"""Create a sorted CommentTree from a flat list of Comments.""" """Create a sorted CommentTree from a flat list of Comments."""
self.tree: List[CommentInTree] = []
self.tree: list[CommentInTree] = []
self.sort = sort self.sort = sort
self.viewer = viewer self.viewer = viewer
@ -92,7 +93,7 @@ class CommentTree:
self.tree.append(comment) self.tree.append(comment)
@staticmethod @staticmethod
def _sort_tree(tree: List[Comment], sort: CommentTreeSortOption) -> List[Comment]:
def _sort_tree(tree: list[Comment], sort: CommentTreeSortOption) -> list[Comment]:
"""Sort the tree by the desired ordering (recursively). """Sort the tree by the desired ordering (recursively).
Because Python's sorted() function is stable, the ordering of any comments that Because Python's sorted() function is stable, the ordering of any comments that
@ -118,7 +119,7 @@ class CommentTree:
return tree return tree
@staticmethod @staticmethod
def _prune_empty_branches(tree: Sequence[Comment]) -> List[Comment]:
def _prune_empty_branches(tree: Sequence[Comment]) -> list[Comment]:
"""Remove branches from the tree with no visible comments.""" """Remove branches from the tree with no visible comments."""
pruned_tree = [] pruned_tree = []
@ -269,7 +270,7 @@ class CommentInTree(ObjectProxy):
super().__init__(comment) super().__init__(comment)
self.collapsed_state: Optional[str] = None self.collapsed_state: Optional[str] = None
self.replies: List[CommentInTree] = []
self.replies: list[CommentInTree] = []
self.has_visible_descendant = False self.has_visible_descendant = False
self.num_children = 0 self.num_children = 0
self.depth = 0 self.depth = 0
@ -308,7 +309,7 @@ class CommentInTree(ObjectProxy):
reply.recursively_collapse() reply.recursively_collapse()
@property @property
def relevance_sorting_value(self) -> Tuple[int, ...]:
def relevance_sorting_value(self) -> tuple[int, ...]:
"""Value to use for the comment with the "relevance" comment sorting method. """Value to use for the comment with the "relevance" comment sorting method.
Returns a tuple, which allows sorting the comments into "tiers" and then still Returns a tuple, which allows sorting the comments into "tiers" and then still

6
tildes/tildes/models/database_model.py

@ -4,7 +4,7 @@
"""Contains the base DatabaseModel class.""" """Contains the base DatabaseModel class."""
from datetime import timedelta from datetime import timedelta
from typing import Any, Optional, Type, TypeVar
from typing import Any, Optional, TypeVar
from marshmallow import Schema from marshmallow import Schema
from sqlalchemy import event from sqlalchemy import event
@ -28,7 +28,7 @@ NAMING_CONVENTION = {
def attach_set_listener( def attach_set_listener(
class_: Type["DatabaseModelBase"], attribute: str, instance: "DatabaseModelBase"
class_: type["DatabaseModelBase"], attribute: str, instance: "DatabaseModelBase"
) -> None: ) -> None:
"""Attach the SQLAlchemy ORM "set" attribute listener.""" """Attach the SQLAlchemy ORM "set" attribute listener."""
# pylint: disable=unused-argument # pylint: disable=unused-argument
@ -48,7 +48,7 @@ class DatabaseModelBase:
# declare the type of __table__ so mypy understands it when checking __eq__ # declare the type of __table__ so mypy understands it when checking __eq__
__table__: Table __table__: Table
schema_class: Optional[Type[Schema]] = None
schema_class: Optional[type[Schema]] = None
def __eq__(self, other: Any) -> bool: def __eq__(self, other: Any) -> bool:
"""Equality comparison method - check if primary key values match.""" """Equality comparison method - check if primary key values match."""

8
tildes/tildes/models/group/group.py

@ -4,7 +4,7 @@
"""Contains the Group class.""" """Contains the Group class."""
from datetime import datetime from datetime import datetime
from typing import List, Optional
from typing import Optional
from pyramid.security import Allow, Authenticated, Deny, DENY_ALL, Everyone from pyramid.security import Allow, Authenticated, Deny, DENY_ALL, Everyone
from sqlalchemy import ( from sqlalchemy import (
@ -67,8 +67,8 @@ class Group(DatabaseModel):
is_user_treated_as_topic_source: bool = Column( is_user_treated_as_topic_source: bool = Column(
Boolean, nullable=False, server_default="false" Boolean, nullable=False, server_default="false"
) )
common_topic_tags: List[str] = Column(TagList, nullable=False, server_default="{}")
important_topic_tags: List[str] = Column(
common_topic_tags: list[str] = Column(TagList, nullable=False, server_default="{}")
important_topic_tags: list[str] = Column(
TagList, nullable=False, server_default="{}" TagList, nullable=False, server_default="{}"
) )
@ -96,7 +96,7 @@ class Group(DatabaseModel):
self.sidebar_rendered_html = None self.sidebar_rendered_html = None
@property @property
def autocomplete_topic_tags(self) -> List[str]:
def autocomplete_topic_tags(self) -> list[str]:
"""Return the topic tags that should be offered as autocomplete options.""" """Return the topic tags that should be offered as autocomplete options."""
global_options = ["nsfw", "spoiler", "coronaviruses.covid19"] global_options = ["nsfw", "spoiler", "coronaviruses.covid19"]

4
tildes/tildes/models/group/group_wiki_page.py

@ -5,7 +5,7 @@
from datetime import datetime from datetime import datetime
from pathlib import Path, PurePath from pathlib import Path, PurePath
from typing import List, Optional
from typing import Optional
from pygit2 import Repository, Signature from pygit2 import Repository, Signature
from pyramid.security import Allow, DENY_ALL, Everyone from pyramid.security import Allow, DENY_ALL, Everyone
@ -104,7 +104,7 @@ class GroupWikiPage(DatabaseModel):
return self.file_path.stem return self.file_path.stem
@property @property
def folders(self) -> List[PurePath]:
def folders(self) -> list[PurePath]:
"""Return a list of the folders the page is inside (if any).""" """Return a list of the folders the page is inside (if any)."""
path = PurePath(self.path) path = PurePath(self.path)

8
tildes/tildes/models/log/log.py

@ -3,7 +3,7 @@
"""Contains the Log class.""" """Contains the Log class."""
from typing import Any, Dict, Optional
from typing import Any, Optional
from pyramid.request import Request from pyramid.request import Request
from sqlalchemy import BigInteger, Column, event, ForeignKey, Table, TIMESTAMP from sqlalchemy import BigInteger, Column, event, ForeignKey, Table, TIMESTAMP
@ -75,7 +75,7 @@ class Log(DatabaseModel, BaseLog):
self, self,
event_type: LogEventType, event_type: LogEventType,
request: Request, request: Request,
info: Optional[Dict[str, Any]] = None,
info: Optional[dict[str, Any]] = None,
): ):
"""Create a new log entry. """Create a new log entry.
@ -106,7 +106,7 @@ class LogComment(DatabaseModel, BaseLog):
event_type: LogEventType, event_type: LogEventType,
request: Request, request: Request,
comment: Comment, comment: Comment,
info: Optional[Dict[str, Any]] = None,
info: Optional[dict[str, Any]] = None,
): ):
"""Create a new log entry related to a specific comment.""" """Create a new log entry related to a specific comment."""
# pylint: disable=non-parent-init-called # pylint: disable=non-parent-init-called
@ -135,7 +135,7 @@ class LogTopic(DatabaseModel, BaseLog):
event_type: LogEventType, event_type: LogEventType,
request: Request, request: Request,
topic: Topic, topic: Topic,
info: Optional[Dict[str, Any]] = None,
info: Optional[dict[str, Any]] = None,
): ):
"""Create a new log entry related to a specific topic.""" """Create a new log entry related to a specific topic."""
# pylint: disable=non-parent-init-called # pylint: disable=non-parent-init-called

5
tildes/tildes/models/message/message.py

@ -12,8 +12,9 @@ This might feel a bit unusual since it splits "all messages" across two tables/c
but it simplifies a lot of things when organizing them into threads. but it simplifies a lot of things when organizing them into threads.
""" """
from collections.abc import Sequence
from datetime import datetime from datetime import datetime
from typing import List, Optional, Sequence
from typing import Optional
from pyramid.security import Allow, DENY_ALL from pyramid.security import Allow, DENY_ALL
from sqlalchemy import ( from sqlalchemy import (
@ -91,7 +92,7 @@ class MessageConversation(DatabaseModel):
# is dangerous and *will* break if user_id values ever get larger than integers # is dangerous and *will* break if user_id values ever get larger than integers
# can hold. I'm comfortable doing something that will only be an issue if the site # can hold. I'm comfortable doing something that will only be an issue if the site
# reaches 2.1 billion users though, I think this would be the least of the problems. # reaches 2.1 billion users though, I think this would be the least of the problems.
unread_user_ids: List[int] = Column(
unread_user_ids: list[int] = Column(
ARRAY(Integer), nullable=False, server_default="{}" ARRAY(Integer), nullable=False, server_default="{}"
) )

3
tildes/tildes/models/model_query.py

@ -5,7 +5,8 @@
# pylint: disable=self-cls-assignment # pylint: disable=self-cls-assignment
from __future__ import annotations from __future__ import annotations
from typing import Any, Iterator, TypeVar
from collections.abc import Iterator
from typing import Any, TypeVar
from pyramid.request import Request from pyramid.request import Request
from sqlalchemy import event from sqlalchemy import event

9
tildes/tildes/models/pagination.py

@ -4,8 +4,9 @@
"""Contains the PaginatedQuery and PaginatedResults classes.""" """Contains the PaginatedQuery and PaginatedResults classes."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Iterator, Sequence
from itertools import chain from itertools import chain
from typing import Any, Iterator, List, Optional, Sequence, TypeVar
from typing import Any, Optional, TypeVar
from pyramid.request import Request from pyramid.request import Request
from sqlalchemy import Column, func, inspect from sqlalchemy import Column, func, inspect
@ -39,12 +40,12 @@ class PaginatedQuery(ModelQuery):
if not self.is_reversed: if not self.is_reversed:
return super().__iter__() return super().__iter__()
results: List[ModelType] = list(super().__iter__())
results: list[ModelType] = list(super().__iter__())
return iter(reversed(results)) return iter(reversed(results))
@property @property
def sorting_columns(self) -> List[Column]:
def sorting_columns(self) -> list[Column]:
"""Return the columns being used for sorting.""" """Return the columns being used for sorting."""
if not self._sort_column: if not self._sort_column:
raise AttributeError raise AttributeError
@ -56,7 +57,7 @@ class PaginatedQuery(ModelQuery):
return [self._sort_column] return [self._sort_column]
@property @property
def sorting_columns_desc(self) -> List[Column]:
def sorting_columns_desc(self) -> list[Column]:
"""Return descending versions of the sorting columns.""" """Return descending versions of the sorting columns."""
return [col.desc() for col in self.sorting_columns] return [col.desc() for col in self.sorting_columns]

15
tildes/tildes/models/topic/topic.py

@ -4,10 +4,11 @@
"""Contains the Topic class.""" """Contains the Topic class."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Iterable
from datetime import datetime, timedelta from datetime import datetime, timedelta
from itertools import chain from itertools import chain
from pathlib import PurePosixPath from pathlib import PurePosixPath
from typing import Any, Dict, Iterable, List, Optional, TYPE_CHECKING
from typing import Any, Optional, TYPE_CHECKING
from urllib.parse import urlparse from urllib.parse import urlparse
from pyramid.security import Allow, Authenticated, Deny, DENY_ALL, Everyone from pyramid.security import Allow, Authenticated, Deny, DENY_ALL, Everyone
@ -122,7 +123,7 @@ class Topic(DatabaseModel):
_markdown: Optional[str] = deferred(Column("markdown", Text)) _markdown: Optional[str] = deferred(Column("markdown", Text))
rendered_html: Optional[str] = Column(Text) rendered_html: Optional[str] = Column(Text)
link: Optional[str] = Column(Text) link: Optional[str] = Column(Text)
content_metadata: Dict[str, Any] = Column(
content_metadata: dict[str, Any] = Column(
MutableDict.as_mutable(JSONB(none_as_null=True)) MutableDict.as_mutable(JSONB(none_as_null=True))
) )
num_comments: int = Column(Integer, nullable=False, server_default="0") num_comments: int = Column(Integer, nullable=False, server_default="0")
@ -130,7 +131,7 @@ class Topic(DatabaseModel):
_is_voting_closed: bool = Column( _is_voting_closed: bool = Column(
"is_voting_closed", Boolean, nullable=False, server_default="false", index=True "is_voting_closed", Boolean, nullable=False, server_default="false", index=True
) )
tags: List[str] = Column(TagList, nullable=False, server_default="{}")
tags: list[str] = Column(TagList, nullable=False, server_default="{}")
is_official: bool = Column(Boolean, nullable=False, server_default="false") is_official: bool = Column(Boolean, nullable=False, server_default="false")
is_locked: bool = Column(Boolean, nullable=False, server_default="false") is_locked: bool = Column(Boolean, nullable=False, server_default="false")
search_tsv: Any = deferred(Column(TSVECTOR)) search_tsv: Any = deferred(Column(TSVECTOR))
@ -184,7 +185,7 @@ class Topic(DatabaseModel):
self.last_edited_time = utc_now() self.last_edited_time = utc_now()
@property @property
def important_tags(self) -> List[str]:
def important_tags(self) -> list[str]:
"""Return only the topic's "important" tags.""" """Return only the topic's "important" tags."""
global_important_tags = ["nsfw", "spoiler"] global_important_tags = ["nsfw", "spoiler"]
@ -200,7 +201,7 @@ class Topic(DatabaseModel):
] ]
@property @property
def unimportant_tags(self) -> List[str]:
def unimportant_tags(self) -> list[str]:
"""Return only the topic's tags that *aren't* considered "important".""" """Return only the topic's tags that *aren't* considered "important"."""
important_tags = set(self.important_tags) important_tags = set(self.important_tags)
return [tag for tag in self.tags if tag not in important_tags] return [tag for tag in self.tags if tag not in important_tags]
@ -552,7 +553,7 @@ class Topic(DatabaseModel):
return self.content_metadata.get(key) return self.content_metadata.get(key)
@property @property
def content_metadata_for_display(self) -> List[str]:
def content_metadata_for_display(self) -> list[str]:
"""Return a list of the content's metadata strings, suitable for display.""" """Return a list of the content's metadata strings, suitable for display."""
if not self.content_type: if not self.content_type:
return [] return []
@ -582,7 +583,7 @@ class Topic(DatabaseModel):
return metadata_strings return metadata_strings
@property @property
def content_metadata_fields_for_display(self) -> Dict[str, str]:
def content_metadata_fields_for_display(self) -> dict[str, str]:
"""Return a dict of the metadata fields and values, suitable for display.""" """Return a dict of the metadata fields and values, suitable for display."""
if not self.content_metadata: if not self.content_metadata:
return {} return {}

3
tildes/tildes/models/topic/topic_query.py

@ -4,7 +4,8 @@
"""Contains the TopicQuery class.""" """Contains the TopicQuery class."""
from __future__ import annotations from __future__ import annotations
from typing import Any, Sequence
from collections.abc import Sequence
from typing import Any
from pyramid.request import Request from pyramid.request import Request
from sqlalchemy import func from sqlalchemy import func

6
tildes/tildes/models/topic/topic_schedule.py

@ -4,7 +4,7 @@
"""Contains the TopicSchedule class.""" """Contains the TopicSchedule class."""
from datetime import datetime from datetime import datetime
from typing import List, Optional
from typing import Optional
from dateutil.rrule import rrule from dateutil.rrule import rrule
from jinja2.sandbox import SandboxedEnvironment from jinja2.sandbox import SandboxedEnvironment
@ -58,7 +58,7 @@ class TopicSchedule(DatabaseModel):
nullable=False, nullable=False,
) )
markdown: str = Column(Text, nullable=False) markdown: str = Column(Text, nullable=False)
tags: List[str] = Column(TagList, nullable=False, server_default="{}")
tags: list[str] = Column(TagList, nullable=False, server_default="{}")
next_post_time: Optional[datetime] = Column( next_post_time: Optional[datetime] = Column(
TIMESTAMP(timezone=True), nullable=True, index=True TIMESTAMP(timezone=True), nullable=True, index=True
) )
@ -79,7 +79,7 @@ class TopicSchedule(DatabaseModel):
group: Group, group: Group,
title: str, title: str,
markdown: str, markdown: str,
tags: List[str],
tags: list[str],
next_post_time: datetime, next_post_time: datetime,
recurrence_rule: Optional[rrule] = None, recurrence_rule: Optional[rrule] = None,
user: Optional[User] = None, user: Optional[User] = None,

8
tildes/tildes/models/user/user.py

@ -4,7 +4,7 @@
"""Contains the User class.""" """Contains the User class."""
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import List, NoReturn, Optional
from typing import NoReturn, Optional
from pyotp import TOTP from pyotp import TOTP
from pyramid.security import ( from pyramid.security import (
@ -84,7 +84,7 @@ class User(DatabaseModel):
) )
two_factor_enabled: bool = Column(Boolean, nullable=False, server_default="false") two_factor_enabled: bool = Column(Boolean, nullable=False, server_default="false")
two_factor_secret: Optional[str] = deferred(Column(Text)) two_factor_secret: Optional[str] = deferred(Column(Text))
two_factor_backup_codes: List[str] = deferred(Column(ARRAY(Text)))
two_factor_backup_codes: list[str] = deferred(Column(ARRAY(Text)))
created_time: datetime = Column( created_time: datetime = Column(
TIMESTAMP(timezone=True), TIMESTAMP(timezone=True),
nullable=False, nullable=False,
@ -130,7 +130,7 @@ class User(DatabaseModel):
ban_expiry_time: Optional[datetime] = Column(TIMESTAMP(timezone=True)) ban_expiry_time: Optional[datetime] = Column(TIMESTAMP(timezone=True))
home_default_order: Optional[TopicSortOption] = Column(ENUM(TopicSortOption)) home_default_order: Optional[TopicSortOption] = Column(ENUM(TopicSortOption))
home_default_period: Optional[str] = Column(Text) home_default_period: Optional[str] = Column(Text)
filtered_topic_tags: List[str] = Column(
filtered_topic_tags: list[str] = Column(
TagList, nullable=False, server_default="{}" TagList, nullable=False, server_default="{}"
) )
comment_label_weight: Optional[float] = Column(REAL) comment_label_weight: Optional[float] = Column(REAL)
@ -322,7 +322,7 @@ class User(DatabaseModel):
return self.num_unread_messages + self.num_unread_notifications return self.num_unread_messages + self.num_unread_notifications
@property @property
def auth_principals(self) -> List[str]:
def auth_principals(self) -> list[str]:
"""Return the user's authorization principals (used for permissions).""" """Return the user's authorization principals (used for permissions)."""
principals = [permission.auth_principal for permission in self.permissions] principals = [permission.auth_principal for permission in self.permissions]

10
tildes/tildes/request_methods.py

@ -3,7 +3,7 @@
"""Define and attach request methods to the Pyramid request object.""" """Define and attach request methods to the Pyramid request object."""
from typing import Any, Dict, Optional, Tuple
from typing import Any, Optional
from pyramid.config import Configurator from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPTooManyRequests from pyramid.httpexceptions import HTTPTooManyRequests
@ -114,7 +114,7 @@ def apply_rate_limit(request: Request, action_name: str) -> None:
def current_listing_base_url( def current_listing_base_url(
request: Request, query: Optional[Dict[str, Any]] = None
request: Request, query: Optional[dict[str, Any]] = None
) -> str: ) -> str:
"""Return the "base" url for the current listing route. """Return the "base" url for the current listing route.
@ -123,7 +123,7 @@ def current_listing_base_url(
The `query` argument allows adding query variables to the generated url. The `query` argument allows adding query variables to the generated url.
""" """
base_vars_by_route: Dict[str, Tuple[str, ...]] = {
base_vars_by_route: dict[str, tuple[str, ...]] = {
"bookmarks": ("per_page", "type"), "bookmarks": ("per_page", "type"),
"group": ("order", "period", "per_page", "tag", "unfiltered"), "group": ("order", "period", "per_page", "tag", "unfiltered"),
"group_search": ("order", "period", "per_page", "q"), "group_search": ("order", "period", "per_page", "q"),
@ -149,7 +149,7 @@ def current_listing_base_url(
def current_listing_normal_url( def current_listing_normal_url(
request: Request, query: Optional[Dict[str, Any]] = None
request: Request, query: Optional[dict[str, Any]] = None
) -> str: ) -> str:
"""Return the "normal" url for the current listing route. """Return the "normal" url for the current listing route.
@ -158,7 +158,7 @@ def current_listing_normal_url(
The `query` argument allows adding query variables to the generated url. The `query` argument allows adding query variables to the generated url.
""" """
normal_vars_by_route: Dict[str, Tuple[str, ...]] = {
normal_vars_by_route: dict[str, tuple[str, ...]] = {
"bookmarks": ("order", "period", "per_page"), "bookmarks": ("order", "period", "per_page"),
"votes": ("order", "period", "per_page"), "votes": ("order", "period", "per_page"),
"group": ("order", "period", "per_page"), "group": ("order", "period", "per_page"),

7
tildes/tildes/schemas/fields.py

@ -5,7 +5,8 @@
import enum import enum
import re import re
from typing import Any, Mapping, Optional, Type
from collections.abc import Mapping
from typing import Any, Optional
import sqlalchemy_utils import sqlalchemy_utils
from marshmallow.exceptions import ValidationError from marshmallow.exceptions import ValidationError
@ -24,7 +25,9 @@ DataType = Optional[Mapping[str, Any]]
class Enum(Field): class Enum(Field):
"""Field for a native Python Enum (or subclasses).""" """Field for a native Python Enum (or subclasses)."""
def __init__(self, enum_class: Optional[Type] = None, *args: Any, **kwargs: Any):
def __init__(
self, enum_class: Optional[type[enum.Enum]] = None, *args: Any, **kwargs: Any
):
"""Initialize the field with an optional enum class.""" """Initialize the field with an optional enum class."""
# pylint: disable=keyword-arg-before-vararg # pylint: disable=keyword-arg-before-vararg
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)

5
tildes/tildes/schemas/topic.py

@ -4,7 +4,6 @@
"""Validation/dumping schema for topics.""" """Validation/dumping schema for topics."""
import re import re
import typing
from typing import Any from typing import Any
from urllib.parse import urlparse from urllib.parse import urlparse
@ -65,7 +64,7 @@ class TopicSchema(Schema):
new_data = data.copy() new_data = data.copy()
tags: typing.List[str] = []
tags: list[str] = []
for tag in new_data["tags"]: for tag in new_data["tags"]:
tag = tag.lower() tag = tag.lower()
@ -99,7 +98,7 @@ class TopicSchema(Schema):
return new_data return new_data
@validates("tags") @validates("tags")
def validate_tags(self, value: typing.List[str]) -> None:
def validate_tags(self, value: list[str]) -> None:
"""Validate the tags field, raising an error if an issue exists. """Validate the tags field, raising an error if an issue exists.
Note that tags are validated by ensuring that each tag would be a valid group Note that tags are validated by ensuring that each tag would be a valid group

6
tildes/tildes/scrapers/embedly_scraper.py

@ -3,7 +3,7 @@
"""Contains the EmbedlyScraper class.""" """Contains the EmbedlyScraper class."""
from typing import Any, Dict
from typing import Any
from urllib.parse import urlparse from urllib.parse import urlparse
import requests import requests
@ -35,7 +35,7 @@ class EmbedlyScraper:
def scrape_url(self, url: str) -> ScraperResult: def scrape_url(self, url: str) -> ScraperResult:
"""Scrape a url and return the result.""" """Scrape a url and return the result."""
params: Dict[str, Any] = {"key": self.api_key, "format": "json", "url": url}
params: dict[str, Any] = {"key": self.api_key, "format": "json", "url": url}
response = requests.get( response = requests.get(
"https://api.embedly.com/1/extract", params=params, timeout=5 "https://api.embedly.com/1/extract", params=params, timeout=5
@ -45,7 +45,7 @@ class EmbedlyScraper:
return ScraperResult(url, ScraperType.EMBEDLY, response.json()) return ScraperResult(url, ScraperType.EMBEDLY, response.json())
@staticmethod @staticmethod
def get_metadata_from_result(result: ScraperResult) -> Dict[str, Any]:
def get_metadata_from_result(result: ScraperResult) -> dict[str, Any]:
"""Get the metadata that we're interested in out of a scrape result.""" """Get the metadata that we're interested in out of a scrape result."""
if result.scraper_type != ScraperType.EMBEDLY: if result.scraper_type != ScraperType.EMBEDLY:
raise ValueError("Can't process a result from a different scraper.") raise ValueError("Can't process a result from a different scraper.")

6
tildes/tildes/scrapers/youtube_scraper.py

@ -5,7 +5,7 @@
import re import re
from datetime import timedelta from datetime import timedelta
from typing import Any, Dict
from typing import Any
from urllib.parse import parse_qs, urlparse from urllib.parse import parse_qs, urlparse
import requests import requests
@ -59,7 +59,7 @@ class YoutubeScraper:
if not video_id: if not video_id:
raise ValueError("Invalid url, no video ID found.") raise ValueError("Invalid url, no video ID found.")
params: Dict[str, Any] = {
params: dict[str, Any] = {
"key": self.api_key, "key": self.api_key,
"id": video_id, "id": video_id,
"part": "snippet,contentDetails,statistics", "part": "snippet,contentDetails,statistics",
@ -80,7 +80,7 @@ class YoutubeScraper:
return ScraperResult(url, ScraperType.YOUTUBE, video_data) return ScraperResult(url, ScraperType.YOUTUBE, video_data)
@classmethod @classmethod
def get_metadata_from_result(cls, result: ScraperResult) -> Dict[str, Any]:
def get_metadata_from_result(cls, result: ScraperResult) -> dict[str, Any]:
"""Get the metadata that we're interested in out of a scrape result.""" """Get the metadata that we're interested in out of a scrape result."""
if result.scraper_type != ScraperType.YOUTUBE: if result.scraper_type != ScraperType.YOUTUBE:
raise ValueError("Can't process a result from a different scraper.") raise ValueError("Can't process a result from a different scraper.")

2
tildes/tildes/tweens.py

@ -3,8 +3,8 @@
"""Contains Pyramid "tweens", used to insert additional logic into request-handling.""" """Contains Pyramid "tweens", used to insert additional logic into request-handling."""
from collections.abc import Callable
from time import time from time import time
from typing import Callable
from prometheus_client import Histogram from prometheus_client import Histogram
from pyramid.config import Configurator from pyramid.config import Configurator

6
tildes/tildes/typing.py

@ -3,8 +3,8 @@
"""Custom type aliases to use in type annotations.""" """Custom type aliases to use in type annotations."""
from typing import Any, List, Tuple
from typing import Any
# types for an ACE (Access Control Entry), and the ACL (Access Control List) of them # types for an ACE (Access Control Entry), and the ACL (Access Control List) of them
AceType = Tuple[str, Any, str]
AclType = List[AceType]
AceType = tuple[str, Any, str]
AclType = list[AceType]

2
tildes/tildes/views/api/web/exceptions.py

@ -3,7 +3,7 @@
"""Web API exception views.""" """Web API exception views."""
from typing import Sequence
from collections.abc import Sequence
from urllib.parse import quote, urlparse, urlunparse from urllib.parse import quote, urlparse, urlunparse
from marshmallow.exceptions import ValidationError from marshmallow.exceptions import ValidationError

4
tildes/tildes/views/bookmarks.py

@ -1,6 +1,6 @@
"""Views relating to bookmarks.""" """Views relating to bookmarks."""
from typing import Optional, Type, Union
from typing import Optional, Union
from pyramid.request import Request from pyramid.request import Request
from pyramid.view import view_config from pyramid.view import view_config
@ -27,7 +27,7 @@ def get_bookmarks(
# pylint: disable=unused-argument # pylint: disable=unused-argument
user = request.user user = request.user
bookmark_cls: Union[Type[CommentBookmark], Type[TopicBookmark]]
bookmark_cls: Union[type[CommentBookmark], type[TopicBookmark]]
if post_type == "comment": if post_type == "comment":
post_cls = Comment post_cls = Comment

5
tildes/tildes/views/decorators.py

@ -3,7 +3,8 @@
"""Contains decorators for view functions.""" """Contains decorators for view functions."""
from typing import Any, Callable, Dict, Union
from collections.abc import Callable
from typing import Any, Union
from marshmallow import EXCLUDE from marshmallow import EXCLUDE
from marshmallow.fields import Field from marshmallow.fields import Field
@ -15,7 +16,7 @@ from webargs import dict2schema, pyramidparser
def use_kwargs( def use_kwargs(
argmap: Union[Schema, Dict[str, Field]], location: str = "query", **kwargs: Any
argmap: Union[Schema, dict[str, Field]], location: str = "query", **kwargs: Any
) -> Callable: ) -> Callable:
"""Wrap the webargs @use_kwargs decorator with preferred default modifications. """Wrap the webargs @use_kwargs decorator with preferred default modifications.

2
tildes/tildes/views/exceptions.py

@ -3,7 +3,7 @@
"""Views used by Pyramid when an exception is raised.""" """Views used by Pyramid when an exception is raised."""
from typing import Sequence
from collections.abc import Sequence
from urllib.parse import quote_plus from urllib.parse import quote_plus
from marshmallow import ValidationError from marshmallow import ValidationError

6
tildes/tildes/views/financials.py

@ -5,7 +5,7 @@
from collections import defaultdict from collections import defaultdict
from decimal import Decimal from decimal import Decimal
from typing import Dict, List, Optional
from typing import Optional
from pyramid.request import Request from pyramid.request import Request
from pyramid.view import view_config from pyramid.view import view_config
@ -30,7 +30,7 @@ def get_financials(request: Request) -> dict:
) )
# split the entries up by type # split the entries up by type
entries: Dict[str, List] = defaultdict(list)
entries = defaultdict(list)
for entry in financial_entries: for entry in financial_entries:
entries[entry.entry_type.name.lower()].append(entry) entries[entry.entry_type.name.lower()].append(entry)
@ -43,7 +43,7 @@ def get_financials(request: Request) -> dict:
} }
def get_financial_data(db_session: Session) -> Optional[Dict[str, Decimal]]:
def get_financial_data(db_session: Session) -> Optional[dict[str, Decimal]]:
"""Return financial data used to render the donation goal box.""" """Return financial data used to render the donation goal box."""
# get the total sum for each entry type in the financials table relevant to today # get the total sum for each entry type in the financials table relevant to today
financial_totals = ( financial_totals = (

12
tildes/tildes/views/user.py

@ -3,7 +3,7 @@
"""Views related to a specific user.""" """Views related to a specific user."""
from typing import List, Optional, Type, Union
from typing import Optional, Union
from enum import Enum from enum import Enum
from marshmallow.fields import String from marshmallow.fields import String
@ -50,8 +50,8 @@ def get_user(
anchor_type = None anchor_type = None
per_page = 20 per_page = 20
types_to_query: List[Union[Type[Topic], Type[Comment]]]
order_options: Optional[Union[Type[TopicSortOption], Type[CommentSortOption]]]
types_to_query: list[Union[type[Topic], type[Comment]]]
order_options: Optional[Union[type[TopicSortOption], type[CommentSortOption]]]
if post_type == "topic": if post_type == "topic":
types_to_query = [Topic] types_to_query = [Topic]
@ -111,8 +111,8 @@ def get_user_search(
"""Generate the search results page for a user's posts.""" """Generate the search results page for a user's posts."""
user = request.context user = request.context
types_to_query: List[Union[Type[Topic], Type[Comment]]]
order_options: Union[Type[TopicSortOption], Type[CommentSortOption]]
types_to_query: list[Union[type[Topic], type[Comment]]]
order_options: Union[type[TopicSortOption], type[CommentSortOption]]
if post_type == "topic": if post_type == "topic":
types_to_query = [Topic] types_to_query = [Topic]
@ -170,7 +170,7 @@ def get_invite(request: Request) -> dict:
def _get_user_posts( def _get_user_posts(
request: Request, request: Request,
user: User, user: User,
types_to_query: List[Union[Type[Topic], Type[Comment]]],
types_to_query: list[Union[type[Topic], type[Comment]]],
anchor_type: Optional[str], anchor_type: Optional[str],
before: Optional[str], before: Optional[str],
after: Optional[str], after: Optional[str],

4
tildes/tildes/views/votes.py

@ -1,6 +1,6 @@
"""Views relating to voted posts.""" """Views relating to voted posts."""
from typing import Optional, Type, Union
from typing import Optional, Union
from pyramid.request import Request from pyramid.request import Request
from pyramid.view import view_config from pyramid.view import view_config
@ -27,7 +27,7 @@ def get_voted_posts(
# pylint: disable=unused-argument # pylint: disable=unused-argument
user = request.user user = request.user
vote_cls: Union[Type[CommentVote], Type[TopicVote]]
vote_cls: Union[type[CommentVote], type[TopicVote]]
if post_type == "comment": if post_type == "comment":
post_cls = Comment post_cls = Comment

Loading…
Cancel
Save