From 01f7b618fbdc60e203bf04974ec3067c513aee59 Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 15:28:09 +0000 Subject: [PATCH 1/7] chore: init async --- scratchattach/async_api/__init__.py | 0 scratchattach/async_api/site/__init__.py | 0 scratchattach/async_api/site/session.py | 0 3 files changed, 0 insertions(+), 0 deletions(-) create mode 100644 scratchattach/async_api/__init__.py create mode 100644 scratchattach/async_api/site/__init__.py create mode 100644 scratchattach/async_api/site/session.py diff --git a/scratchattach/async_api/__init__.py b/scratchattach/async_api/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/scratchattach/async_api/site/__init__.py b/scratchattach/async_api/site/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/scratchattach/async_api/site/session.py b/scratchattach/async_api/site/session.py new file mode 100644 index 00000000..e69de29b From 7c19b7eba082cd6f9a9e9b6122359598264cbf90 Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 19:07:53 +0000 Subject: [PATCH 2/7] feat: async login --- pyproject.toml | 8 +- scratchattach/async_api/__init__.py | 1 + scratchattach/async_api/site/session.py | 78 ++++ scratchattach/site/session.py | 507 +++++++++++++----------- scratchattach/utils/commons.py | 146 ++++--- tests/util/__init__.py | 18 +- tests/util/keyhandler.py | 6 +- uv.lock | 78 ++++ 8 files changed, 567 insertions(+), 275 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 08cc5d28..6ec94be4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "browser_cookie3", "aiohttp", "rich", + "httpx>=0.28.1", ] readme = "README.md" license = "MIT" @@ -92,4 +93,9 @@ max-complexity = 10 find = { include = ["scratchattach"] } [dependency-groups] -dev = ["cryptography>=46.0.3", "pytest>=9.0.2"] +dev = [ + "cryptography>=46.0.3", + "pytest>=9.0.2", + "pytest-asyncio>=1.3.0", + "python-dotenv>=1.2.1", +] diff --git a/scratchattach/async_api/__init__.py b/scratchattach/async_api/__init__.py index e69de29b..2bd3f8d0 100644 --- a/scratchattach/async_api/__init__.py +++ b/scratchattach/async_api/__init__.py @@ -0,0 +1 @@ +from scratchattach.async_api.site.session import Session, login diff --git a/scratchattach/async_api/site/session.py b/scratchattach/async_api/site/session.py index e69de29b..e994690e 100644 --- a/scratchattach/async_api/site/session.py +++ b/scratchattach/async_api/site/session.py @@ -0,0 +1,78 @@ +import httpx +import re +import contextlib + +from datetime import datetime +from dataclasses import KW_ONLY, dataclass, field +from typing_extensions import Any, AsyncContextManager, Optional, Self + +from scratchattach.utils import commons, exceptions + + +@dataclass +class Session: + _: KW_ONLY + id: str = field(repr=False) + rq: httpx.AsyncClient = field(repr=False) + user_id: int + username: str + xtoken: str + created_at: datetime + + def __post_init__(self): + self.rq.headers["X-Token"] = self.xtoken + + +@contextlib.asynccontextmanager +async def _build_session(*, id: str, rq: httpx.AsyncClient, username: Optional[str]): + try: + data, created_at = commons.decode_session_id(id) + assert data["username"] == username or username is None + # not saving the login ip because it can be considered as a security issue, and is not very helpful + yield Session( + id=id, + rq=rq, + created_at=created_at, + username=data["username"], + user_id=int(data["_auth_user_id"]), + xtoken=data["token"], + ) + finally: + pass + + +async def login( + username: str, password: str, *, client_args: Optional[dict[str, Any]] = None +) -> AsyncContextManager[Session, bool | None]: + if client_args is None: + client_args = {} + + print("TODO: issue_login_warning") + rq = httpx.AsyncClient( + headers=commons.headers.copy() | {"Cookie": "scratchcsrftoken=a;scratchlanguage=en;"}, **client_args + ) + resp = await rq.post("https://scratch.mit.edu/login/", json={"username": username, "password": password}) + if not (match := re.search('"(.*)"', resp.headers.get("Set-Cookie", ""))): + raise exceptions.LoginFailure( + "Either the provided authentication data is wrong or your network is banned from Scratch.\n\nIf you're using an online IDE (like replit.com) Scratch possibly banned its IP address. In this case, try logging in with your session id: https://github.com/TimMcCool/scratchattach/wiki#logging-in" + ) + + session_id = match.group() + + return await login_by_id(session_id, rq=rq) + + +async def login_by_id( + session_id: str, + *, + username: Optional[str] = None, + rq: Optional[httpx.AsyncClient] = None, + client_args: Optional[dict[str, Any]] = None, +) -> AsyncContextManager[Session, bool | None]: + if client_args is None: + client_args = {} + if rq is None: + rq = httpx.AsyncClient(headers=commons.headers.copy(), **client_args) + print("TODO: issue_login_warning") + + return _build_session(id=session_id, rq=rq, username=username) diff --git a/scratchattach/site/session.py b/scratchattach/site/session.py index d22d9597..18901084 100644 --- a/scratchattach/site/session.py +++ b/scratchattach/site/session.py @@ -1,4 +1,5 @@ """Session class and login function""" + from __future__ import annotations import base64 @@ -24,6 +25,7 @@ if TYPE_CHECKING: from _typeshed import FileDescriptorOrPath, SupportsRead from scratchattach.cloud._base import BaseCloud + T = TypeVar("T", bound=BaseCloud) else: T = TypeVar("T") @@ -33,6 +35,7 @@ from . import activity, classroom, forum, studio, user, project, backpack_asset, alert from . import typed_dicts + # noinspection PyProtectedMember from ._base import BaseSiteComponent from scratchattach.cloud import cloud, _base @@ -45,6 +48,7 @@ ratelimit_cache: dict[str, list[float]] = {} + def enforce_ratelimit(__type: str, name: str, amount: int = 5, duration: int = 60) -> None: cache = ratelimit_cache cache.setdefault(__type, []) @@ -61,8 +65,10 @@ def enforce_ratelimit(__type: str, name: str, amount: int = 5, duration: int = 6 "Don't spam-create studios or similar, it WILL get you banned." ) + C = TypeVar("C", bound=BaseSiteComponent) + @dataclass class Session(BaseSiteComponent): """ @@ -77,6 +83,7 @@ class Session(BaseSiteComponent): mute_status: Information about commenting restrictions of the associated account banned: Returns True if the associated account is banned """ + username: str = field(repr=False, default="") _user: Optional[user.User] = field(repr=False, default=None) @@ -113,8 +120,8 @@ def __rich__(self): warnings.warn(f"Ignored KeyError: {e}") ret = Table( - f"[link={self.connect_linked_user().url}]{escape(self.username)}[/]", - f"Created: {self.time_created}", expand=True) + f"[link={self.connect_linked_user().url}]{escape(self.username)}[/]", f"Created: {self.time_created}", expand=True + ) ret.add_row("Email", escape(str(self.email))) ret.add_row("Language", escape(str(self.language))) @@ -158,7 +165,7 @@ def _update_from_dict(self, data: Union[dict, typed_dicts.SessionDict]): data = cast(typed_dicts.SessionDict, data) - self.xtoken = data['user']['token'] + self.xtoken = data["user"]["token"] self._headers["X-Token"] = self.xtoken self.has_outstanding_email_confirmation = data["flags"]["has_outstanding_email_confirmation"] @@ -175,17 +182,19 @@ def _update_from_dict(self, data: Union[dict, typed_dicts.SessionDict]): self.banned = data["user"]["banned"] if self.banned: - warnings.warn(f"Warning: The account {self.username} you logged in to is BANNED. " - f"Some features may not work properly.") + warnings.warn( + f"Warning: The account {self.username} you logged in to is BANNED. Some features may not work properly." + ) if self.has_outstanding_email_confirmation: - warnings.warn(f"Warning: The account {self.username} you logged is not email confirmed. " - f"Some features may not work properly.") + warnings.warn( + f"Warning: The account {self.username} you logged is not email confirmed. Some features may not work properly." + ) return True def _process_session_id(self): assert self.id - data, self.time_created = decode_session_id(self.id) + data, self.time_created = commons.decode_session_id(self.id) self.username = data["username"] # if self._user: @@ -204,7 +213,9 @@ def _process_session_id(self): def _assert_ocular_auth(self): if not self.ocular_token: - raise ValueError(f"No ocular token supplied for {self}! You can add one by using Session.set_ocular_token(YOUR_TOKEN).") + raise ValueError( + f"No ocular token supplied for {self}! You can add one by using Session.set_ocular_token(YOUR_TOKEN)." + ) def set_ocular_token(self, token: str): self.ocular_token = token @@ -241,9 +252,12 @@ def set_country(self, country: str = "Antarctica"): Arguments: country (str): The country to relocate to """ - requests.post("https://scratch.mit.edu/accounts/settings/", - data={"country": country}, - headers=self._headers, cookies=self._cookies) + requests.post( + "https://scratch.mit.edu/accounts/settings/", + data={"country": country}, + headers=self._headers, + cookies=self._cookies, + ) def resend_email(self, password: str): """ @@ -252,10 +266,12 @@ def resend_email(self, password: str): Keyword arguments: password (str): Password associated with the session (not stored) """ - requests.post("https://scratch.mit.edu/accounts/email_change/", - data={"email_address": self.get_new_email_address(), - "password": password}, - headers=self._headers, cookies=self._cookies) + requests.post( + "https://scratch.mit.edu/accounts/email_change/", + data={"email_address": self.get_new_email_address(), "password": password}, + headers=self._headers, + cookies=self._cookies, + ) @property @deprecated("Use get_new_email_address instead.") @@ -277,8 +293,7 @@ def get_new_email_address(self) -> str: Returns: str: The email that this session wants to switch to """ - response = requests.get("https://scratch.mit.edu/accounts/email_change/", - headers=self._headers, cookies=self._cookies) + response = requests.get("https://scratch.mit.edu/accounts/email_change/", headers=self._headers, cookies=self._cookies) soup = BeautifulSoup(response.content, "html.parser") @@ -300,17 +315,16 @@ def logout(self): """ Sends a logout request to scratch. (Might not do anything, might log out this account on other ips/sessions.) """ - requests.post("https://scratch.mit.edu/accounts/logout/", - headers=self._headers, cookies=self._cookies) + requests.post("https://scratch.mit.edu/accounts/logout/", headers=self._headers, cookies=self._cookies) @property def ocular_headers(self) -> dict[str, str]: self._assert_ocular_auth() return { "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36", + "(KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36", "referer": "https://ocular.jeffalo.net/", - "authorization": self.ocular_token + "authorization": self.ocular_token, } def get_ocular_status(self) -> typed_dicts.OcularUserDict: @@ -324,13 +338,11 @@ def get_ocular_status(self) -> typed_dicts.OcularUserDict: def set_ocular_status(self, status: Optional[str] = None, color: Optional[str] = None) -> None: self._assert_ocular_auth() old = self.get_ocular_status() - payload = {"color": color or old["color"], - "status": status or old["status"]} + payload = {"color": color or old["color"], "status": status or old["status"]} - assert requests.put(f"https://my-ocular.jeffalo.net/api/user/{old["name"]}", - json=payload, headers=self.ocular_headers).json() == { - "ok": "user updated" - }, f"Error occured on setting ocular status. auth/me response: {old}" + assert requests.put( + f"https://my-ocular.jeffalo.net/api/user/{old['name']}", json=payload, headers=self.ocular_headers + ).json() == {"ok": "user updated"}, f"Error occured on setting ocular status. auth/me response: {old}" def messages(self, *, limit: int = 40, offset: int = 0, date_limit=None, filter_by=None) -> list[activity.Activity]: """ @@ -351,7 +363,11 @@ def messages(self, *, limit: int = 40, offset: int = 0, date_limit=None, filter_ data = commons.api_iterative( f"https://api.scratch.mit.edu/users/{self._username}/messages", - limit=limit, offset=offset, _headers=self._headers, cookies=self._cookies, add_params=add_params + limit=limit, + offset=offset, + _headers=self._headers, + cookies=self._cookies, + add_params=add_params, ) return commons.parse_object_list(data, activity.Activity, self) @@ -361,11 +377,15 @@ def admin_messages(self, *, limit=40, offset=0) -> list[dict]: """ return commons.api_iterative( f"https://api.scratch.mit.edu/users/{self._username}/messages/admin", - limit=limit, offset=offset, _headers=self._headers, cookies=self._cookies + limit=limit, + offset=offset, + _headers=self._headers, + cookies=self._cookies, ) - def classroom_alerts(self, _classroom: Optional[classroom.Classroom | int] = None, mode: str = "Last created", - page: Optional[int] = None): + def classroom_alerts( + self, _classroom: Optional[classroom.Classroom | int] = None, mode: str = "Last created", page: Optional[int] = None + ): """ Load and parse admin alerts, optionally for a specific class, using https://scratch.mit.edu/site-api/classrooms/alerts/ @@ -377,15 +397,18 @@ def classroom_alerts(self, _classroom: Optional[classroom.Classroom | int] = Non _classroom = _classroom.id if _classroom is None: - _classroom_str = '' + _classroom_str = "" else: _classroom_str = f"{_classroom}/" ascsort, descsort = get_class_sort_mode(mode) - data = requests.get(f"https://scratch.mit.edu/site-api/classrooms/alerts/{_classroom_str}", - params={"page": page, "ascsort": ascsort, "descsort": descsort}, - headers=self._headers, cookies=self._cookies).json() + data = requests.get( + f"https://scratch.mit.edu/site-api/classrooms/alerts/{_classroom_str}", + params={"page": page, "ascsort": ascsort, "descsort": descsort}, + headers=self._headers, + cookies=self._cookies, + ).json() alerts = [alert.EducatorAlert.from_json(alert_data, self) for alert_data in data] @@ -409,12 +432,14 @@ def message_count(self) -> int: Returns: int: message count """ - return json.loads(requests.get( - f"https://scratch.mit.edu/messages/ajax/get-message-count/", - headers=self._headers, - cookies=self._cookies, - timeout=10, - ).text)["msg_count"] + return json.loads( + requests.get( + f"https://scratch.mit.edu/messages/ajax/get-message-count/", + headers=self._headers, + cookies=self._cookies, + timeout=10, + ).text + )["msg_count"] # Front-page-related stuff: @@ -430,7 +455,11 @@ def feed(self, *, limit=20, offset=0, date_limit=None) -> list[activity.Activity add_params = f"&dateLimit={date_limit}" data = commons.api_iterative( f"https://api.scratch.mit.edu/users/{self._username}/following/users/activity", - limit=limit, offset=offset, _headers=self._headers, cookies=self._cookies, add_params=add_params + limit=limit, + offset=offset, + _headers=self._headers, + cookies=self._cookies, + add_params=add_params, ) return commons.parse_object_list(data, activity.Activity, self) @@ -448,7 +477,10 @@ def loved_by_followed_users(self, *, limit=40, offset=0) -> list[project.Project """ data = commons.api_iterative( f"https://api.scratch.mit.edu/users/{self._username}/following/users/loves", - limit=limit, offset=offset, _headers=self._headers, cookies=self._cookies + limit=limit, + offset=offset, + _headers=self._headers, + cookies=self._cookies, ) return commons.parse_object_list(data, project.Project, self) @@ -464,15 +496,20 @@ def shared_by_followed_users(self, *, limit=40, offset=0) -> list[project.Projec """ data = commons.api_iterative( f"https://api.scratch.mit.edu/users/{self._username}/following/users/projects", - limit = limit, offset = offset, _headers = self._headers, cookies = self._cookies + limit=limit, + offset=offset, + _headers=self._headers, + cookies=self._cookies, ) ret = commons.parse_object_list(data, project.Project, self) if not ret: - warnings.warn(f"`shared_by_followed_users` got empty list `[]`. Note that this method is not supported for " - f"accounts made after 2018.") + warnings.warn( + f"`shared_by_followed_users` got empty list `[]`. Note that this method is not supported for " + f"accounts made after 2018." + ) return ret - def in_followed_studios(self, *, limit=40, offset=0) -> list['project.Project']: + def in_followed_studios(self, *, limit=40, offset=0) -> list["project.Project"]: """ Returns the "Projects in studios I'm following" section (frontpage). This section is only visible to old accounts (until ~2018) @@ -484,18 +521,25 @@ def in_followed_studios(self, *, limit=40, offset=0) -> list['project.Project']: """ data = commons.api_iterative( f"https://api.scratch.mit.edu/users/{self._username}/following/studios/projects", - limit = limit, offset = offset, _headers=self._headers, cookies = self._cookies + limit=limit, + offset=offset, + _headers=self._headers, + cookies=self._cookies, ) ret = commons.parse_object_list(data, project.Project, self) if not ret: - warnings.warn(f"`in_followed_studios` got empty list `[]`. Note that this method is not supported for " - f"accounts made after 2018.") + warnings.warn( + f"`in_followed_studios` got empty list `[]`. Note that this method is not supported for " + f"accounts made after 2018." + ) return ret # -- Project JSON editing capabilities --- # These are set to staticmethods right now, but they probably should not be def connect_empty_project_pb(self) -> editor.Project: - pb = editor.Project.from_json(empty_project_json) # in the future, ideally just init a new editor.Project, instead of loading an empty one + pb = editor.Project.from_json( + empty_project_json + ) # in the future, ideally just init a new editor.Project, instead of loading an empty one pb._session = self return pb @@ -522,11 +566,7 @@ def download_asset(asset_id_with_file_ext, *, filename: Optional[str] = None, fp ) open(f"{fp}{filename}", "wb").write(response.content) except Exception: - raise ( - exceptions.FetchError( - "Failed to download asset" - ) - ) + raise (exceptions.FetchError("Failed to download asset")) def upload_asset(self, asset_content, *, asset_id=None, file_ext=None): data = asset_content if isinstance(asset_content, bytes) else open(asset_content, "rb").read() @@ -548,8 +588,9 @@ def upload_asset(self, asset_content, *, asset_id=None, file_ext=None): # --- Search --- - def search_projects(self, *, query: str = "", mode: str = "trending", language: str = "en", limit: int = 40, - offset: int = 0) -> list[project.Project]: + def search_projects( + self, *, query: str = "", mode: str = "trending", language: str = "en", limit: int = 40, offset: int = 0 + ) -> list[project.Project]: """ Uses the Scratch search to search projects. @@ -567,12 +608,16 @@ def search_projects(self, *, query: str = "", mode: str = "trending", language: query = f"&q={query}" if query else "" response = commons.api_iterative( - f"https://api.scratch.mit.edu/search/projects", limit=limit, offset=offset, - add_params=f"&language={language}&mode={mode}{query}") + f"https://api.scratch.mit.edu/search/projects", + limit=limit, + offset=offset, + add_params=f"&language={language}&mode={mode}{query}", + ) return commons.parse_object_list(response, project.Project, self) - def explore_projects(self, *, query: str = "*", mode: str = "trending", language: str = "en", limit: int = 40, - offset: int = 0) -> list[project.Project]: + def explore_projects( + self, *, query: str = "*", mode: str = "trending", language: str = "en", limit: int = 40, offset: int = 0 + ) -> list[project.Project]: """ Gets projects from the explore page. @@ -590,31 +635,43 @@ def explore_projects(self, *, query: str = "*", mode: str = "trending", language list: List that contains the explore page projects. """ response = commons.api_iterative( - f"https://api.scratch.mit.edu/explore/projects", limit=limit, offset=offset, - add_params=f"&language={language}&mode={mode}&q={query}") + f"https://api.scratch.mit.edu/explore/projects", + limit=limit, + offset=offset, + add_params=f"&language={language}&mode={mode}&q={query}", + ) return commons.parse_object_list(response, project.Project, self) - def search_studios(self, *, query: str = "", mode: str = "trending", language: str = "en", limit: int = 40, - offset: int = 0) -> list[studio.Studio]: + def search_studios( + self, *, query: str = "", mode: str = "trending", language: str = "en", limit: int = 40, offset: int = 0 + ) -> list[studio.Studio]: query = f"&q={query}" if query else "" response = commons.api_iterative( - f"https://api.scratch.mit.edu/search/studios", limit=limit, offset=offset, - add_params=f"&language={language}&mode={mode}{query}") + f"https://api.scratch.mit.edu/search/studios", + limit=limit, + offset=offset, + add_params=f"&language={language}&mode={mode}{query}", + ) return commons.parse_object_list(response, studio.Studio, self) - def explore_studios(self, *, query: str = "", mode: str = "trending", language: str = "en", limit: int = 40, - offset: int = 0) -> list[studio.Studio]: + def explore_studios( + self, *, query: str = "", mode: str = "trending", language: str = "en", limit: int = 40, offset: int = 0 + ) -> list[studio.Studio]: query = f"&q={query}" if query else "" response = commons.api_iterative( - f"https://api.scratch.mit.edu/explore/studios", limit=limit, offset=offset, - add_params=f"&language={language}&mode={mode}{query}") + f"https://api.scratch.mit.edu/explore/studios", + limit=limit, + offset=offset, + add_params=f"&language={language}&mode={mode}{query}", + ) return commons.parse_object_list(response, studio.Studio, self) # --- Create project API --- - def create_project(self, *, title: Optional[str] = None, project_json: dict = empty_project_json, - parent_id=None) -> project.Project: # not working + def create_project( + self, *, title: Optional[str] = None, project_json: dict = empty_project_json, parent_id=None + ) -> project.Project: # not working """ Creates a project on the Scratch website. @@ -625,16 +682,17 @@ def create_project(self, *, title: Optional[str] = None, project_json: dict = em enforce_ratelimit("create_scratch_project", "creating Scratch projects") if title is None: - title = f'Untitled-{random.randint(0, 1<<16)}' + title = f"Untitled-{random.randint(0, 1 << 16)}" params = { - 'is_remix': '0' if parent_id is None else "1", - 'original_id': parent_id, - 'title': title, + "is_remix": "0" if parent_id is None else "1", + "original_id": parent_id, + "title": title, } - response = requests.post('https://projects.scratch.mit.edu/', params=params, cookies=self._cookies, - headers=self._headers, json=project_json).json() + response = requests.post( + "https://projects.scratch.mit.edu/", params=params, cookies=self._cookies, headers=self._headers, json=project_json + ).json() return self.connect_project(response["content-name"]) def create_studio(self, *, title: Optional[str] = None, description: Optional[str] = None) -> studio.Studio: @@ -650,8 +708,7 @@ def create_studio(self, *, title: Optional[str] = None, description: Optional[st if self.new_scratcher: raise exceptions.Unauthorized(f"\nNew scratchers (like {self.username}) cannot create studios.") - response = requests.post("https://scratch.mit.edu/studios/create/", - cookies=self._cookies, headers=self._headers) + response = requests.post("https://scratch.mit.edu/studios/create/", cookies=self._cookies, headers=self._headers) studio_id = webscrape_count(response.json()["redirect"], "/studios/", "/") new_studio = self.connect_studio(studio_id) @@ -663,7 +720,7 @@ def create_studio(self, *, title: Optional[str] = None, description: Optional[st return new_studio - def create_class(self, title: str, desc: str = '') -> classroom.Classroom: + def create_class(self, title: str, desc: str = "") -> classroom.Classroom: """ Create a class on the scratch website @@ -676,17 +733,21 @@ def create_class(self, title: str, desc: str = '') -> classroom.Classroom: if not self.is_teacher: raise exceptions.Unauthorized(f"{self.username} is not a teacher; can't create class") - data = requests.post("https://scratch.mit.edu/classes/create_classroom/", - json={"title": title, "description": desc}, - headers=self._headers, cookies=self._cookies).json() + data = requests.post( + "https://scratch.mit.edu/classes/create_classroom/", + json={"title": title, "description": desc}, + headers=self._headers, + cookies=self._cookies, + ).json() class_id = data[0]["id"] return self.connect_classroom(class_id) # --- My stuff page --- - def mystuff_projects(self, filter_arg: str = "all", *, page: int = 1, sort_by: str = '', descending: bool = True) \ - -> list[project.Project]: + def mystuff_projects( + self, filter_arg: str = "all", *, page: int = 1, sort_by: str = "", descending: bool = True + ) -> list[project.Project]: """ Gets the projects from the "My stuff" page. @@ -716,26 +777,33 @@ def mystuff_projects(self, filter_arg: str = "all", *, page: int = 1, sort_by: s ).json() projects = [] for target in targets: - projects.append(project.Project( - id=target["pk"], _session=self, author_name=self._username, - comments_allowed=None, instructions=None, notes=None, - created=target["fields"]["datetime_created"], - last_modified=target["fields"]["datetime_modified"], - share_date=target["fields"]["datetime_shared"], - thumbnail_url="https:" + target["fields"]["thumbnail_url"], - favorites=target["fields"]["favorite_count"], - loves=target["fields"]["love_count"], - remixes=target["fields"]["remixers_count"], - views=target["fields"]["view_count"], - title=target["fields"]["title"], - comment_count=target["fields"]["commenters_count"] - )) + projects.append( + project.Project( + id=target["pk"], + _session=self, + author_name=self._username, + comments_allowed=None, + instructions=None, + notes=None, + created=target["fields"]["datetime_created"], + last_modified=target["fields"]["datetime_modified"], + share_date=target["fields"]["datetime_shared"], + thumbnail_url="https:" + target["fields"]["thumbnail_url"], + favorites=target["fields"]["favorite_count"], + loves=target["fields"]["love_count"], + remixes=target["fields"]["remixers_count"], + views=target["fields"]["view_count"], + title=target["fields"]["title"], + comment_count=target["fields"]["commenters_count"], + ) + ) return projects except Exception: raise exceptions.FetchError() - def mystuff_studios(self, filter_arg: str = "all", *, page: int = 1, sort_by: str = "", descending: bool = True) \ - -> list[studio.Studio]: + def mystuff_studios( + self, filter_arg: str = "all", *, page: int = 1, sort_by: str = "", descending: bool = True + ) -> list[studio.Studio]: if descending: ascsort = "" descsort = sort_by @@ -749,24 +817,29 @@ def mystuff_studios(self, filter_arg: str = "all", *, page: int = 1, sort_by: st params=params, headers=headers, cookies=self._cookies, - timeout=10 + timeout=10, ).json() studios = [] for target in targets: - studios.append(studio.Studio( - id=target["pk"], _session=self, - title=target["fields"]["title"], - description=None, - host_id=target["fields"]["owner"]["pk"], - host_name=target["fields"]["owner"]["username"], - open_to_all=None, comments_allowed=None, - image_url="https:" + target["fields"]["thumbnail_url"], - created=target["fields"]["datetime_created"], - modified=target["fields"]["datetime_modified"], - follower_count=None, manager_count=None, - curator_count=target["fields"]["curators_count"], - project_count=target["fields"]["projecters_count"] - )) + studios.append( + studio.Studio( + id=target["pk"], + _session=self, + title=target["fields"]["title"], + description=None, + host_id=target["fields"]["owner"]["pk"], + host_name=target["fields"]["owner"]["username"], + open_to_all=None, + comments_allowed=None, + image_url="https:" + target["fields"]["thumbnail_url"], + created=target["fields"]["datetime_created"], + modified=target["fields"]["datetime_modified"], + follower_count=None, + manager_count=None, + curator_count=target["fields"]["curators_count"], + project_count=target["fields"]["projecters_count"], + ) + ) return studios except Exception: raise exceptions.FetchError() @@ -779,21 +852,26 @@ def mystuff_classes(self, mode: str = "Last created", page: Optional[int] = None raise exceptions.Unauthorized(f"{self.username} is not a teacher; can't have classes") ascsort, descsort = get_class_sort_mode(mode) - classes_data = requests.get("https://scratch.mit.edu/site-api/classrooms/all/", - params={"page": page, "ascsort": ascsort, "descsort": descsort}, - headers=self._headers, cookies=self._cookies).json() + classes_data = requests.get( + "https://scratch.mit.edu/site-api/classrooms/all/", + params={"page": page, "ascsort": ascsort, "descsort": descsort}, + headers=self._headers, + cookies=self._cookies, + ).json() classes = [] for data in classes_data: fields = data["fields"] educator_pf = fields["educator_profile"] - classes.append(classroom.Classroom( - id=data["pk"], - title=fields["title"], - classtoken=fields["token"], - datetime=datetime.datetime.fromisoformat(fields["datetime_created"]), - author=user.User( - username=educator_pf["user"]["username"], id=educator_pf["user"]["pk"], _session=self), - _session=self)) + classes.append( + classroom.Classroom( + id=data["pk"], + title=fields["title"], + classtoken=fields["token"], + datetime=datetime.datetime.fromisoformat(fields["datetime_created"]), + author=user.User(username=educator_pf["user"]["username"], id=educator_pf["user"]["pk"], _session=self), + _session=self, + ) + ) return classes def mystuff_ended_classes(self, mode: str = "Last created", page: Optional[int] = None) -> list[classroom.Classroom]: @@ -801,21 +879,26 @@ def mystuff_ended_classes(self, mode: str = "Last created", page: Optional[int] raise exceptions.Unauthorized(f"{self.username} is not a teacher; can't have (deleted) classes") ascsort, descsort = get_class_sort_mode(mode) - classes_data = requests.get("https://scratch.mit.edu/site-api/classrooms/closed/", - params={"page": page, "ascsort": ascsort, "descsort": descsort}, - headers=self._headers, cookies=self._cookies).json() + classes_data = requests.get( + "https://scratch.mit.edu/site-api/classrooms/closed/", + params={"page": page, "ascsort": ascsort, "descsort": descsort}, + headers=self._headers, + cookies=self._cookies, + ).json() classes = [] for data in classes_data: fields = data["fields"] educator_pf = fields["educator_profile"] - classes.append(classroom.Classroom( - id=data["pk"], - title=fields["title"], - classtoken=fields["token"], - datetime=datetime.datetime.fromisoformat(fields["datetime_created"]), - author=user.User( - username=educator_pf["user"]["username"], id=educator_pf["user"]["pk"], _session=self), - _session=self)) + classes.append( + classroom.Classroom( + id=data["pk"], + title=fields["title"], + classtoken=fields["token"], + datetime=datetime.datetime.fromisoformat(fields["datetime_created"]), + author=user.User(username=educator_pf["user"]["username"], id=educator_pf["user"]["pk"], _session=self), + _session=self, + ) + ) return classes def backpack(self, limit: int = 20, offset: int = 0) -> list[backpack_asset.BackpackAsset]: @@ -826,8 +909,7 @@ def backpack(self, limit: int = 20, offset: int = 0) -> list[backpack_asset.Back list: List that contains the backpack items """ data = commons.api_iterative( - f"https://backpack.scratch.mit.edu/{self._username}", - limit=limit, offset=offset, _headers=self._headers + f"https://backpack.scratch.mit.edu/{self._username}", limit=limit, offset=offset, _headers=self._headers ) return commons.parse_object_list(data, backpack_asset.BackpackAsset, self) @@ -845,12 +927,12 @@ def become_scratcher_invite(self) -> dict: If you are a new Scratcher and have been invited for becoming a Scratcher, this API endpoint will provide more info on the invite. """ - return requests.get(f"https://api.scratch.mit.edu/users/{self.username}/invites", headers=self._headers, - cookies=self._cookies).json() + return requests.get( + f"https://api.scratch.mit.edu/users/{self.username}/invites", headers=self._headers, cookies=self._cookies + ).json() # --- Connect classes inheriting from BaseCloud --- - @overload def connect_cloud(self, project_id, *, cloud_class: type[T]) -> T: """ @@ -858,7 +940,7 @@ def connect_cloud(self, project_id, *, cloud_class: type[T]) -> T: Args: project_id: - + Keyword arguments: CloudClass: The class that the returned object should be of. By default, this class is scratchattach.cloud.ScratchCloud. @@ -873,16 +955,16 @@ def connect_cloud(self, project_id) -> cloud.ScratchCloud: Args: project_id: - + Keyword arguments: CloudClass: The class that the returned object should be of. By default, this class is scratchattach.cloud.ScratchCloud. Returns: Type[scratchattach.cloud._base.BaseCloud]: An object representing the cloud of a project. Can be of any class inheriting from BaseCloud. """ + # noinspection PyPep8Naming - def connect_cloud(self, project_id, *, cloud_class: Optional[type[_base.BaseCloud]] = None) \ - -> _base.BaseCloud: + def connect_cloud(self, project_id, *, cloud_class: Optional[type[_base.BaseCloud]] = None) -> _base.BaseCloud: cloud_class = cloud_class or cloud.ScratchCloud return cloud_class(project_id=project_id, _session=self) @@ -893,21 +975,22 @@ def connect_scratch_cloud(self, project_id) -> cloud.ScratchCloud: """ return cloud.ScratchCloud(project_id=project_id, _session=self) - def connect_tw_cloud(self, project_id, *, purpose="", contact="", - cloud_host="wss://clouddata.turbowarp.org") -> cloud.TwCloud: + def connect_tw_cloud( + self, project_id, *, purpose="", contact="", cloud_host="wss://clouddata.turbowarp.org" + ) -> cloud.TwCloud: """ Returns: scratchattach.cloud.TwCloud: An object representing the TurboWarp cloud of a project. """ - return cloud.TwCloud(project_id=project_id, purpose=purpose, contact=contact, cloud_host=cloud_host, - _session=self) + return cloud.TwCloud(project_id=project_id, purpose=purpose, contact=contact, cloud_host=cloud_host, _session=self) # --- Connect classes inheriting from BaseSiteComponent --- # noinspection PyPep8Naming # Class is camelcase here - def _make_linked_object(self, identificator_name, identificator, __class: type[C], - NotFoundException: type[Exception]) -> C: + def _make_linked_object( + self, identificator_name, identificator, __class: type[C], NotFoundException: type[Exception] + ) -> C: """ The Session class doesn't save the login in a ._session attribute, but IS the login ITSELF. @@ -946,7 +1029,8 @@ def find_username_from_id(self, user_id: int) -> str: comment = you.post_comment("scratchattach", commentee_id=int(user_id)) except exceptions.CommentPostFailure: raise exceptions.BadRequest( - "After posting a comment, you need to wait 10 seconds before you can connect users by id again.") + "After posting a comment, you need to wait 10 seconds before you can connect users by id again." + ) except exceptions.BadRequest: raise exceptions.UserNotFound("Invalid user id") except Exception as e: @@ -979,18 +1063,17 @@ def connect_user_by_id(self, user_id: int) -> user.User: scratchattach.user.User: An object that represents the requested user and allows you to perform actions on the user (like user.follow) """ # noinspection PyDeprecation - return self._make_linked_object("username", self.find_username_from_id(user_id), user.User, - exceptions.UserNotFound) + return self._make_linked_object("username", self.find_username_from_id(user_id), user.User, exceptions.UserNotFound) def connect_project(self, project_id) -> project.Project: """ - Gets a project using this session, connects the session to the Project object to allow authenticated actions -sess - Args: - project_id (int): ID of the requested project + Gets a project using this session, connects the session to the Project object to allow authenticated actions + sess + Args: + project_id (int): ID of the requested project - Returns: - scratchattach.project.Project: An object that represents the requested project and allows you to perform actions on the project (like project.love) + Returns: + scratchattach.project.Project: An object that represents the requested project and allows you to perform actions on the project (like project.love) """ return self._make_linked_object("id", int(project_id), project.Project, exceptions.ProjectNotFound) @@ -1028,8 +1111,7 @@ def connect_classroom_from_token(self, class_token) -> classroom.Classroom: Returns: scratchattach.classroom.Classroom: An object representing the requested classroom """ - return self._make_linked_object("classtoken", int(class_token), classroom.Classroom, - exceptions.ClassroomNotFound) + return self._make_linked_object("classtoken", int(class_token), classroom.Classroom, exceptions.ClassroomNotFound) def connect_topic(self, topic_id) -> forum.ForumTopic: """ @@ -1045,7 +1127,6 @@ def connect_topic(self, topic_id) -> forum.ForumTopic: return self._make_linked_object("id", int(topic_id), forum.ForumTopic, exceptions.ForumContentNotFound) def connect_topic_list(self, category_id, *, page=1): - """ Gets the topics from a forum category. Data web-scraped from Scratch's forums UI. Data is up-to-date. @@ -1061,28 +1142,29 @@ def connect_topic_list(self, category_id, *, page=1): """ try: - response = requests.get(f"https://scratch.mit.edu/discuss/{category_id}/?page={page}", - headers=self._headers, cookies=self._cookies) - soup = BeautifulSoup(response.content, 'html.parser') + response = requests.get( + f"https://scratch.mit.edu/discuss/{category_id}/?page={page}", headers=self._headers, cookies=self._cookies + ) + soup = BeautifulSoup(response.content, "html.parser") except Exception as e: raise exceptions.FetchError(str(e)) try: - category_name = soup.find('h4').find("span").get_text() + category_name = soup.find("h4").find("span").get_text() except Exception: raise exceptions.BadRequest("Invalid category id") try: - topics = soup.find_all('tr') + topics = soup.find_all("tr") topics.pop(0) return_topics = [] for topic in topics: - title_link = topic.find('a') + title_link = topic.find("a") title = title_link.text.strip() - topic_id = title_link['href'].split('/')[-2] + topic_id = title_link["href"].split("/")[-2] - columns = topic.find_all('td') + columns = topic.find_all("td") columns = [column.text for column in columns] if len(columns) == 1: # This is a sticky topic -> Skip it @@ -1091,9 +1173,16 @@ def connect_topic_list(self, category_id, *, page=1): last_updated = columns[3].split(" ")[0] + " " + columns[3].split(" ")[1] return_topics.append( - forum.ForumTopic(_session=self, id=int(topic_id), title=title, category_name=category_name, - last_updated=last_updated, reply_count=int(columns[1]), - view_count=int(columns[2]))) + forum.ForumTopic( + _session=self, + id=int(topic_id), + title=title, + category_name=category_name, + last_updated=last_updated, + reply_count=int(columns[1]), + view_count=int(columns[2]), + ) + ) return return_topics except Exception as e: raise exceptions.ScrapeError(str(e)) @@ -1108,8 +1197,7 @@ def connect_featured(self) -> other_apis.FeaturedData: def connect_message_events(self, *, update_interval=2) -> message_events.MessageEvents: # shortcut for connect_linked_user().message_events() - return message_events.MessageEvents(user.User(username=self.username, _session=self), - update_interval=update_interval) + return message_events.MessageEvents(user.User(username=self.username, _session=self), update_interval=update_interval) def connect_filterbot(self, *, log_deletions=True) -> filterbot.Filterbot: return filterbot.Filterbot(user.User(username=self.username, _session=self), log_deletions=log_deletions) @@ -1125,40 +1213,6 @@ def get_cookies(self) -> dict[str, str]: return self._cookies -# ------ # - -def decode_session_id(session_id: str) -> tuple[dict[str, str], datetime.datetime]: - """ - Extract the JSON data from the main part of a session ID string - Session id is in the format: - :: - - p1 contains a base64 JSON string (if it starts with `.`, then it is zlib compressed) - p2 is a base 62 encoded timestamp - p3 might be a `synchronous signature` for the first 2 parts (might be useless for us) - - The dict has these attributes: - - username - - _auth_user_id - - testcookie - - _auth_user_backend - - token - - login-ip - - _language - - django_timezone - - _auth_user_hash - """ - p1, p2, _ = session_id.split(':') - p1_bytes = base64.urlsafe_b64decode(p1 + "==") - if p1.startswith('".'): - p1_bytes = zlib.decompress(p1_bytes) - - return ( - json.loads(p1_bytes), - datetime.datetime.fromtimestamp(commons.b62_decode(p2)) - ) - - # ------ # suppressed_login_warning = local() @@ -1188,7 +1242,7 @@ def issue_login_warning() -> None: "then make sure to EITHER instead load them from environment variables or files OR remember to remove them before " "you share your code with anyone else. If you want to remove this warning, " "use `warnings.filterwarnings('ignore', category=scratchattach.LoginDataWarning)`", - exceptions.LoginDataWarning + exceptions.LoginDataWarning, ) @@ -1250,8 +1304,10 @@ def login(username, password, *, timeout=10) -> Session: _headers["Cookie"] = "scratchcsrftoken=a;scratchlanguage=en;" with requests.no_error_handling(): request = requests.post( - "https://scratch.mit.edu/login/", json={"username": username, "password": password}, headers=_headers, - timeout=timeout + "https://scratch.mit.edu/login/", + json={"username": username, "password": password}, + headers=_headers, + timeout=timeout, ) try: @@ -1260,7 +1316,8 @@ def login(username, password, *, timeout=10) -> Session: session_id = str(result.group()) except Exception: raise exceptions.LoginFailure( - "Either the provided authentication data is wrong or your network is banned from Scratch.\n\nIf you're using an online IDE (like replit.com) Scratch possibly banned its IP address. In this case, try logging in with your session id: https://github.com/TimMcCool/scratchattach/wiki#logging-in") + "Either the provided authentication data is wrong or your network is banned from Scratch.\n\nIf you're using an online IDE (like replit.com) Scratch possibly banned its IP address. In this case, try logging in with your session id: https://github.com/TimMcCool/scratchattach/wiki#logging-in" + ) # Create session object: with suppress_login_warning(): @@ -1277,15 +1334,17 @@ def login_by_session_string(session_string: str) -> Session: try: assert session_data.get("id") with suppress_login_warning(): - return login_by_id(session_data["id"], username=session_data.get("username"), - password=session_data.get("password")) + return login_by_id( + session_data["id"], username=session_data.get("username"), password=session_data.get("password") + ) except Exception: pass try: assert session_data.get("session_id") with suppress_login_warning(): - return login_by_id(session_data["session_id"], username=session_data.get("username"), - password=session_data.get("password")) + return login_by_id( + session_data["session_id"], username=session_data.get("username"), password=session_data.get("password") + ) except Exception: pass try: diff --git a/scratchattach/utils/commons.py b/scratchattach/utils/commons.py index a24eef0c..439948fe 100644 --- a/scratchattach/utils/commons.py +++ b/scratchattach/utils/commons.py @@ -1,8 +1,13 @@ """v2 ready: Common functions used by various internal modules""" + from __future__ import annotations import string +import base64 +import json +import zlib +from datetime import datetime from typing import Optional, Final, Any, TypeVar, Callable, TYPE_CHECKING, Union, overload from threading import Event as ManualResetEvent from threading import Lock @@ -11,64 +16,65 @@ from .requests import requests from scratchattach.site import _base +from scratchattach.utils import typed_dicts headers: Final = { "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36", + "(KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36", "x-csrftoken": "a", "x-requested-with": "XMLHttpRequest", "referer": "https://scratch.mit.edu", } empty_project_json: Final = { - 'targets': [ + "targets": [ { - 'isStage': True, - 'name': 'Stage', - 'variables': { - '`jEk@4|i[#Fk?(8x)AV.-my variable': [ - 'my variable', + "isStage": True, + "name": "Stage", + "variables": { + "`jEk@4|i[#Fk?(8x)AV.-my variable": [ + "my variable", 0, ], }, - 'lists': {}, - 'broadcasts': {}, - 'blocks': {}, - 'comments': {}, - 'currentCostume': 0, - 'costumes': [ + "lists": {}, + "broadcasts": {}, + "blocks": {}, + "comments": {}, + "currentCostume": 0, + "costumes": [ { - 'name': '', - 'bitmapResolution': 1, - 'dataFormat': 'svg', - 'assetId': '14e46ec3e2ba471c2adfe8f119052307', - 'md5ext': '14e46ec3e2ba471c2adfe8f119052307.svg', - 'rotationCenterX': 0, - 'rotationCenterY': 0, + "name": "", + "bitmapResolution": 1, + "dataFormat": "svg", + "assetId": "14e46ec3e2ba471c2adfe8f119052307", + "md5ext": "14e46ec3e2ba471c2adfe8f119052307.svg", + "rotationCenterX": 0, + "rotationCenterY": 0, }, ], - 'sounds': [], - 'volume': 100, - 'layerOrder': 0, - 'tempo': 60, - 'videoTransparency': 50, - 'videoState': 'on', - 'textToSpeechLanguage': None, + "sounds": [], + "volume": 100, + "layerOrder": 0, + "tempo": 60, + "videoTransparency": 50, + "videoState": "on", + "textToSpeechLanguage": None, }, ], - 'monitors': [], - 'extensions': [], - 'meta': { - 'semver': '3.0.0', - 'vm': '2.3.0', - 'agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) ' - 'Chrome/124.0.0.0 Safari/537.36', + "monitors": [], + "extensions": [], + "meta": { + "semver": "3.0.0", + "vm": "2.3.0", + "agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", }, } -def api_iterative_data(fetch_func: Callable[[int, int], list], limit: int, offset: int, max_req_limit: int = 40, - unpack: bool = True) -> list: +def api_iterative_data( + fetch_func: Callable[[int, int], list], limit: int, offset: int, max_req_limit: int = 40, unpack: bool = True +) -> list: """ Iteratively gets data by calling fetch_func with a moving offset and a limit. Once fetch_func returns None, the retrieval is completed. @@ -96,8 +102,16 @@ def api_iterative_data(fetch_func: Callable[[int, int], list], limit: int, offse return api_data -def api_iterative(url: str, *, limit: int, offset: int, max_req_limit: int = 40, add_params: str = "", - _headers: Optional[dict] = None, cookies: Optional[dict] = None): +def api_iterative( + url: str, + *, + limit: int, + offset: int, + max_req_limit: int = 40, + add_params: str = "", + _headers: Optional[dict] = None, + cookies: Optional[dict] = None, +): """ Function for getting data from one of Scratch's iterative JSON API endpoints (like /users//followers, or /users//projects) """ @@ -125,16 +139,16 @@ def fetch(off: int, lim: int): raise exceptions.BadRequest("The passed arguments are invalid") return resp - api_data = api_iterative_data( - fetch, limit, offset, max_req_limit=max_req_limit, unpack=True - ) + api_data = api_iterative_data(fetch, limit, offset, max_req_limit=max_req_limit, unpack=True) return api_data + def _get_object(identificator_name, identificator, __class: type[C], NotFoundException, session=None) -> C: # Internal function: Generalization of the process ran by get_user, get_studio etc. # Builds an object of class that is inheriting from BaseSiteComponent # # Class must inherit from BaseSiteComponent from scratchattach.site import project + try: use_class: type = __class if __class is project.PartialProject: @@ -145,12 +159,12 @@ def _get_object(identificator_name, identificator, __class: type[C], NotFoundExc if r == "429": raise exceptions.Response429( "Your network is blocked or rate-limited by Scratch.\n" - "If you're using an online IDE like replit.com, try running the code on your computer.") + "If you're using an online IDE like replit.com, try running the code on your computer." + ) if not r: # Target is unshared. The cases that this can happen in are hardcoded: if __class is project.PartialProject: # Case: Target is an unshared project. - _object = project.PartialProject(**{identificator_name: identificator, - "shared": False, "_session": session}) + _object = project.PartialProject(**{identificator_name: identificator, "shared": False, "_session": session}) assert isinstance(_object, __class) return _object else: @@ -162,22 +176,28 @@ def _get_object(identificator_name, identificator, __class: type[C], NotFoundExc except Exception as e: raise e + I = TypeVar("I") + + @overload def webscrape_count(raw: str, text_before: str, text_after: str, cls: type[I]) -> I: pass + @overload def webscrape_count(raw: str, text_before: str, text_after: str) -> int: pass -def webscrape_count(raw, text_before, text_after, cls = int): + +def webscrape_count(raw, text_before, text_after, cls=int): return cls(raw.split(text_before)[1].split(text_after)[0]) if TYPE_CHECKING: C = TypeVar("C", bound=_base.BaseSiteComponent) + def parse_object_list(raw, /, __class: type[C], session=None, primary_key="id") -> list[C]: results = [] for raw_dict in raw: @@ -195,9 +215,11 @@ class LockEvent: """ Can be waited on and triggered. Not to be confused with threading.Event, which has to be reset. """ + _event: ManualResetEvent _locks: list[Lock] _access_locks: Lock + def __init__(self): self._event = ManualResetEvent() self._locks = [] @@ -233,12 +255,13 @@ def on(self) -> Lock: lock.acquire(timeout=0) return lock + def get_class_sort_mode(mode: str) -> tuple[str, str]: """ Returns the sort mode for the given mode for classes only """ - ascsort = '' - descsort = '' + ascsort = "" + descsort = "" mode = mode.lower() if mode == "last created": @@ -261,3 +284,32 @@ def b62_decode(s: str): ret = ret * 62 + chars.index(char) return ret + + +def decode_session_id(session_id: str) -> tuple[typed_dicts.SessionIDDict, datetime]: + """ + Extract the JSON data from the main part of a session ID string + Session id is in the format: + :: + + p1 contains a base64 JSON string (if it starts with `.`, then it is zlib compressed) + p2 is a base 62 encoded timestamp + p3 might be a `synchronous signature` for the first 2 parts (might be useless for us) + + The dict has these attributes: + - username + - _auth_user_id + - testcookie + - _auth_user_backend + - token + - login-ip + - _language + - django_timezone + - _auth_user_hash + """ + p1, p2, _ = session_id.split(":") + p1_bytes = base64.urlsafe_b64decode(p1 + "==") + if p1.startswith('".'): + p1_bytes = zlib.decompress(p1_bytes) + + return (json.loads(p1_bytes), datetime.fromtimestamp(b62_decode(p2))) diff --git a/tests/util/__init__.py b/tests/util/__init__.py index fd7f8397..7047eb8d 100644 --- a/tests/util/__init__.py +++ b/tests/util/__init__.py @@ -1,22 +1,26 @@ # utility methods for testing # includes special handlers for authentication etc. +from typing_extensions import AsyncContextManager import warnings from typing import Optional from .keyhandler import get_auth +import scratchattach.async_api as asa from scratchattach import login, Session as _Session, LoginDataWarning -warnings.filterwarnings('ignore', category=LoginDataWarning) +warnings.filterwarnings("ignore", category=LoginDataWarning) _session: Optional[_Session] = None + def credentials_available() -> bool: auth = get_auth() if not auth: return False return auth.get("auth") is not None + def session() -> _Session: global _session @@ -29,7 +33,18 @@ def session() -> _Session: return _session + +async def async_session() -> AsyncContextManager[asa.Session, bool | None]: + auth = get_auth().get("auth") + pw = None if auth is None else auth.get("scratchattachv2") + if pw is None: + raise RuntimeError("Not enough data for login.") + return await asa.login("ScratchAttachV2", pw) + + _teacher_session: Optional[_Session] = None + + def teacher_session() -> Optional[_Session]: global _teacher_session @@ -42,4 +57,3 @@ def teacher_session() -> Optional[_Session]: _teacher_session = login(data["username"], data["password"]) return _teacher_session - diff --git a/tests/util/keyhandler.py b/tests/util/keyhandler.py index 4f076724..0b3d12d4 100644 --- a/tests/util/keyhandler.py +++ b/tests/util/keyhandler.py @@ -4,9 +4,12 @@ from pathlib import Path from typing import Any, Optional, TypeVar +from dotenv import load_dotenv from cryptography.fernet import Fernet from base64 import urlsafe_b64encode +load_dotenv() + def str_2_key(gen: str) -> bytes: if not gen: @@ -61,6 +64,7 @@ def _decrypt_dict(data: dict) -> dict: _cached_auth: Optional[dict[str, Any]] = None + def get_auth() -> dict[str, Any]: try: _auth = _decrypt_dict(tomllib.load(_auth_fp.open("rb"))) @@ -70,5 +74,5 @@ def get_auth() -> dict[str, Any]: _local_auth = tomllib.load(_local_auth_fp.open("rb")) if _local_auth_fp.exists() else {} _cached_auth = _auth | _local_auth - + return _cached_auth diff --git a/uv.lock b/uv.lock index e7ea88d9..2ac52e1d 100644 --- a/uv.lock +++ b/uv.lock @@ -109,6 +109,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fb/76/641ae371508676492379f16e2fa48f4e2c11741bd63c48be4b12a6b09cba/aiosignal-1.4.0-py3-none-any.whl", hash = "sha256:053243f8b92b990551949e63930a839ff0cf0b0ebbe0597b0f3fb19e1a0fe82e", size = 7490, upload-time = "2025-07-03T22:54:42.156Z" }, ] +[[package]] +name = "anyio" +version = "4.12.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "idna" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/96/f0/5eb65b2bb0d09ac6776f2eb54adee6abe8228ea05b20a5ad0e4945de8aac/anyio-4.12.1.tar.gz", hash = "sha256:41cfcc3a4c85d3f05c932da7c26d0201ac36f72abd4435ba90d0464a3ffed703", size = 228685, upload-time = "2026-01-06T11:45:21.246Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/38/0e/27be9fdef66e72d64c0cdc3cc2823101b80585f8119b5c112c2e8f5f7dab/anyio-4.12.1-py3-none-any.whl", hash = "sha256:d405828884fc140aa80a3c667b8beed277f1dfedec42ba031bd6ac3db606ab6c", size = 113592, upload-time = "2026-01-06T11:45:19.497Z" }, +] + [[package]] name = "attrs" version = "25.4.0" @@ -435,6 +448,43 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9a/9a/e35b4a917281c0b8419d4207f4334c8e8c5dbf4f3f5f9ada73958d937dcc/frozenlist-1.8.0-py3-none-any.whl", hash = "sha256:0c18a16eab41e82c295618a77502e17b195883241c563b00f0aa5106fc4eaa0d", size = 13409, upload-time = "2025-10-06T05:38:16.721Z" }, ] +[[package]] +name = "h11" +version = "0.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/01/ee/02a2c011bdab74c6fb3c75474d40b3052059d95df7e73351460c8588d963/h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1", size = 101250, upload-time = "2025-04-24T03:35:25.427Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, +] + +[[package]] +name = "httpcore" +version = "1.0.9" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "certifi" }, + { name = "h11" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, +] + +[[package]] +name = "httpx" +version = "0.28.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "certifi" }, + { name = "httpcore" }, + { name = "idna" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, +] + [[package]] name = "idna" version = "3.11" @@ -866,6 +916,28 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "1.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" }, +] + +[[package]] +name = "python-dotenv" +version = "1.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f0/26/19cadc79a718c5edbec86fd4919a6b6d3f681039a2f6d66d14be94e75fb9/python_dotenv-1.2.1.tar.gz", hash = "sha256:42667e897e16ab0d66954af0e60a9caa94f0fd4ecf3aaf6d2d260eec1aa36ad6", size = 44221, upload-time = "2025-10-26T15:12:10.434Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/14/1b/a298b06749107c305e1fe0f814c6c74aea7b2f1e10989cb30f544a1b3253/python_dotenv-1.2.1-py3-none-any.whl", hash = "sha256:b81ee9561e9ca4004139c6cbba3a238c32b03e4894671e181b671e8cb8425d61", size = 21230, upload-time = "2025-10-26T15:12:09.109Z" }, +] + [[package]] name = "pywin32" version = "311" @@ -931,6 +1003,7 @@ dependencies = [ { name = "aiohttp" }, { name = "browser-cookie3" }, { name = "bs4" }, + { name = "httpx" }, { name = "requests" }, { name = "rich" }, { name = "simplewebsocketserver" }, @@ -950,6 +1023,8 @@ lark = [ dev = [ { name = "cryptography" }, { name = "pytest" }, + { name = "pytest-asyncio" }, + { name = "python-dotenv" }, ] [package.metadata] @@ -957,6 +1032,7 @@ requires-dist = [ { name = "aiohttp" }, { name = "browser-cookie3" }, { name = "bs4" }, + { name = "httpx", specifier = ">=0.28.1" }, { name = "lark", marker = "extra == 'lark'" }, { name = "requests" }, { name = "rich" }, @@ -971,6 +1047,8 @@ provides-extras = ["cli", "lark"] dev = [ { name = "cryptography", specifier = ">=46.0.3" }, { name = "pytest", specifier = ">=9.0.2" }, + { name = "pytest-asyncio", specifier = ">=1.3.0" }, + { name = "python-dotenv", specifier = ">=1.2.1" }, ] [[package]] From aff9006c5e973240dec0a96bf6b306e4bb5bcf81 Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 19:08:28 +0000 Subject: [PATCH 3/7] test: test async login --- tests/test_async_login.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 tests/test_async_login.py diff --git a/tests/test_async_login.py b/tests/test_async_login.py new file mode 100644 index 00000000..62ecf129 --- /dev/null +++ b/tests/test_async_login.py @@ -0,0 +1,18 @@ +import scratchattach.async_api as sa +import asyncio +import util +import warnings +import pytest + + +@pytest.mark.asyncio +async def test_async_login(): + if not util.credentials_available(): + warnings.warn("Skipped test_activity because there were no credentials available.") + return + async with await util.async_session() as sess: + ... + + +if __name__ == "__main__": + asyncio.run(test_async_login()) From e13153f46b4f21355356f96ee3f4e0564adde8c9 Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 19:09:56 +0000 Subject: [PATCH 4/7] feat: typed dicts in utils forgot to add this to previous commit --- scratchattach/utils/typed_dicts.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 scratchattach/utils/typed_dicts.py diff --git a/scratchattach/utils/typed_dicts.py b/scratchattach/utils/typed_dicts.py new file mode 100644 index 00000000..a3dae0e3 --- /dev/null +++ b/scratchattach/utils/typed_dicts.py @@ -0,0 +1,18 @@ +# NOTE: There also exists typed dicts in site/typed_dicts. They should be moved here. + +from typing_extensions import TypedDict + +SessionIDDict = TypedDict( + "SessionIDDict", + { + "username": str, + "_auth_user_id": str, + "testcookie": str, + "_auth_user_backend": str, + "token": str, + "login-ip": str, + "_language": str, + "django_timezone": str, + "_auth_user_hash": str, + }, +) From 051342b083d248368d2ce66b3f9d70aecad9059b Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 20:11:25 +0000 Subject: [PATCH 5/7] feat: export login by id --- scratchattach/async_api/__init__.py | 2 +- tests/test_async_login.py | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/scratchattach/async_api/__init__.py b/scratchattach/async_api/__init__.py index 2bd3f8d0..589d29b9 100644 --- a/scratchattach/async_api/__init__.py +++ b/scratchattach/async_api/__init__.py @@ -1 +1 @@ -from scratchattach.async_api.site.session import Session, login +from scratchattach.async_api.site.session import Session, login, login_by_id diff --git a/tests/test_async_login.py b/tests/test_async_login.py index 62ecf129..6e1256e8 100644 --- a/tests/test_async_login.py +++ b/tests/test_async_login.py @@ -1,4 +1,3 @@ -import scratchattach.async_api as sa import asyncio import util import warnings @@ -11,7 +10,7 @@ async def test_async_login(): warnings.warn("Skipped test_activity because there were no credentials available.") return async with await util.async_session() as sess: - ... + print(sess.user_id) if __name__ == "__main__": From f44038f052cbd200b72d1b0d6a0b2a4ee137e414 Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 20:33:36 +0000 Subject: [PATCH 6/7] feat: session update --- scratchattach/async_api/site/session.py | 70 ++++++++-- scratchattach/utils/typed_dicts.py | 171 +++++++++++++++++++++++- tests/test_async_login.py | 2 + 3 files changed, 234 insertions(+), 9 deletions(-) diff --git a/scratchattach/async_api/site/session.py b/scratchattach/async_api/site/session.py index e994690e..333d9b91 100644 --- a/scratchattach/async_api/site/session.py +++ b/scratchattach/async_api/site/session.py @@ -1,12 +1,13 @@ +import warnings import httpx import re import contextlib from datetime import datetime from dataclasses import KW_ONLY, dataclass, field -from typing_extensions import Any, AsyncContextManager, Optional, Self +from typing_extensions import Any, AsyncContextManager, Optional, Self, cast -from scratchattach.utils import commons, exceptions +from scratchattach.utils import commons, exceptions, typed_dicts @dataclass @@ -19,8 +20,57 @@ class Session: xtoken: str created_at: datetime + # the following attributes are set in the `update()` function. + has_outstanding_email_confirmation: Optional[bool] = None + email: Optional[str] = None + is_new_scratcher: Optional[bool] = None + is_teacher: Optional[bool] = None + is_teacher_invitee: Optional[bool] = None + mute_status: Optional[dict | typed_dicts.SessionOffensesDict] = None + is_banned: Optional[bool] = None + def __post_init__(self): self.rq.headers["X-Token"] = self.xtoken + self.rq.cookies.update( + { + "scratchsessionsid": self.id, + "scratchcsrftoken": "a", + "scratchlanguage": "en", + "accept": "application/json", + "Content-Type": "application/json", + } + ) + + def __str__(self) -> str: + return f"-L {self.username}" + + async def update(self): + # I don't really see the point of abstracting the update url and stuff + resp = await self.rq.post("https://scratch.mit.edu/session") + data = cast(typed_dicts.SessionDict, resp.json()) + self.has_outstanding_email_confirmation = data["flags"]["has_outstanding_email_confirmation"] + + self.email = data["user"]["email"] + + self.is_new_scratcher = data["permissions"]["new_scratcher"] + self.is_teacher = data["permissions"]["educator"] + self.is_teacher_invitee = data["permissions"]["educator_invitee"] + + self.mute_status = data["permissions"]["mute_status"] + + self.username = data["user"]["username"] + self.banned = data["user"]["banned"] + + if self.xtoken != data["user"]["token"]: + warnings.warn(f"Differing xtoken {data['user']['token']!r}") + if self.banned: + warnings.warn( + f"Warning: The account {self.username} you logged in to is BANNED. Some features may not work properly." + ) + if self.has_outstanding_email_confirmation: + warnings.warn( + f"Warning: The account {self.username} you logged is not email confirmed. Some features may not work properly." + ) @contextlib.asynccontextmanager @@ -41,6 +91,14 @@ async def _build_session(*, id: str, rq: httpx.AsyncClient, username: Optional[s pass +def _make_rq(kwargs: Optional[dict[str, Any]]) -> httpx.AsyncClient: + if kwargs is None: + kwargs = {} + return httpx.AsyncClient( + follow_redirects=True, headers=commons.headers | {"Cookie": "scratchcsrftoken=a;scratchlanguage=en;"}, **kwargs + ) + + async def login( username: str, password: str, *, client_args: Optional[dict[str, Any]] = None ) -> AsyncContextManager[Session, bool | None]: @@ -48,9 +106,7 @@ async def login( client_args = {} print("TODO: issue_login_warning") - rq = httpx.AsyncClient( - headers=commons.headers.copy() | {"Cookie": "scratchcsrftoken=a;scratchlanguage=en;"}, **client_args - ) + rq = _make_rq(client_args) resp = await rq.post("https://scratch.mit.edu/login/", json={"username": username, "password": password}) if not (match := re.search('"(.*)"', resp.headers.get("Set-Cookie", ""))): raise exceptions.LoginFailure( @@ -69,10 +125,8 @@ async def login_by_id( rq: Optional[httpx.AsyncClient] = None, client_args: Optional[dict[str, Any]] = None, ) -> AsyncContextManager[Session, bool | None]: - if client_args is None: - client_args = {} if rq is None: - rq = httpx.AsyncClient(headers=commons.headers.copy(), **client_args) + rq = _make_rq(client_args) print("TODO: issue_login_warning") return _build_session(id=session_id, rq=rq, username=username) diff --git a/scratchattach/utils/typed_dicts.py b/scratchattach/utils/typed_dicts.py index a3dae0e3..5b9b6f5d 100644 --- a/scratchattach/utils/typed_dicts.py +++ b/scratchattach/utils/typed_dicts.py @@ -1,6 +1,10 @@ +from __future__ import annotations + +from scratchattach.cloud import _base +from typing_extensions import TypedDict, Union, Optional, Required, NotRequired, OrderedDict + # NOTE: There also exists typed dicts in site/typed_dicts. They should be moved here. -from typing_extensions import TypedDict SessionIDDict = TypedDict( "SessionIDDict", @@ -16,3 +20,168 @@ "_auth_user_hash": str, }, ) + + +class SessionUserDict(TypedDict): + id: int + banned: bool + should_vpn: bool + username: str + token: str + thumbnailUrl: str + dateJoined: str + email: str + birthYear: int + birthMonth: int + gender: str + + +class SessionOffenseDict(TypedDict): + expiresAt: float + messageType: str + createdAt: float + + +class SessionOffensesDict(TypedDict): + offenses: list[SessionOffenseDict] + showWarning: bool + muteExpiresAt: float + currentMessageType: str + + +class SessionPermissionsDict(TypedDict): + admin: bool + scratcher: bool + new_scratcher: bool + invited_scratcher: bool + social: bool + educator: bool + educator_invitee: bool + student: bool + mute_status: Union[dict, SessionOffensesDict] + + +class SessionFlagsDict(TypedDict): + must_reset_password: bool + must_complete_registration: bool + has_outstanding_email_confirmation: bool + show_welcome: bool + confirm_email_banner: bool + unsupported_browser_banner: bool + with_parent_email: bool + project_comments_enabled: bool + gallery_comments_enabled: bool + userprofile_comments_enabled: bool + everything_is_totally_normal: bool + + +class SessionDict(TypedDict): + user: SessionUserDict + permissions: SessionPermissionsDict + flags: SessionFlagsDict + + +class OcularUserMetaDict(TypedDict): + updated: str + updatedBy: str + + +class OcularUserDict(TypedDict): + _id: str + name: str + status: str + color: str + meta: OcularUserMetaDict + + +class UserHistoryDict(TypedDict): + joined: str + + +class UserProfileDict(TypedDict): + id: int + status: str + bio: str + country: str + images: dict[str, str] + membership_label: NotRequired[int] + membership_avatar_badge: NotRequired[int] + + +class UserDict(TypedDict): + id: NotRequired[int] + username: NotRequired[str] + scratchteam: NotRequired[bool] + history: NotRequired[UserHistoryDict] + profile: NotRequired[UserProfileDict] + + +class CloudLogActivityDict(TypedDict): + user: str + verb: str + name: str + value: Union[str, float, int] + timestamp: int + cloud: _base.AnyCloud + + +class CloudActivityDict(TypedDict): + method: str + name: str + value: Union[str, float, int] + project_id: int + cloud: _base.AnyCloud + + +class ClassroomDict(TypedDict): + id: int + title: str + description: str + status: str + date_start: NotRequired[str] + date_end: NotRequired[Optional[str]] + images: NotRequired[dict[str, str]] + educator: UserDict + is_closed: NotRequired[bool] + + +class StudioHistoryDict(TypedDict): + created: str + modified: str + + +class StudioStatsDict(TypedDict): + followers: int + managers: int + projects: int + + +class StudioDict(TypedDict): + id: int + title: str + description: str + host: int + open_to_all: bool + comments_allowed: bool + image: str + history: StudioHistoryDict + stats: NotRequired[StudioStatsDict] + + +class StudioRoleDict(TypedDict): + manager: bool + curator: bool + invited: bool + following: bool + + +class PlaceholderProjectDataMetadataDict(TypedDict): + title: str + description: str + + +# https://github.com/GarboMuffin/placeholder/blob/e1e98953342a40abbd626a111f621711f74e783b/src/routes/projects/%5Bproject%5D/%2Bpage.server.ts#L19 +class PlaceholderProjectDataDict(TypedDict): + metadata: PlaceholderProjectDataMetadataDict + md5extsToSha256: OrderedDict[str, str] + adminOwnershipToken: Optional[str] diff --git a/tests/test_async_login.py b/tests/test_async_login.py index 6e1256e8..7c6076f9 100644 --- a/tests/test_async_login.py +++ b/tests/test_async_login.py @@ -11,6 +11,8 @@ async def test_async_login(): return async with await util.async_session() as sess: print(sess.user_id) + await sess.update() + print(repr(sess)) if __name__ == "__main__": From e1c414ccf903d57bc7192a64a2109dd45f30ada1 Mon Sep 17 00:00:00 2001 From: rtk <107722825+faretek1@users.noreply.github.com> Date: Sat, 31 Jan 2026 20:41:05 +0000 Subject: [PATCH 7/7] feat: hide some attrs on repr --- scratchattach/async_api/site/session.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scratchattach/async_api/site/session.py b/scratchattach/async_api/site/session.py index 333d9b91..137b476e 100644 --- a/scratchattach/async_api/site/session.py +++ b/scratchattach/async_api/site/session.py @@ -17,12 +17,12 @@ class Session: rq: httpx.AsyncClient = field(repr=False) user_id: int username: str - xtoken: str + xtoken: str = field(repr=False) created_at: datetime # the following attributes are set in the `update()` function. has_outstanding_email_confirmation: Optional[bool] = None - email: Optional[str] = None + email: Optional[str] = field(repr=False, default=None) is_new_scratcher: Optional[bool] = None is_teacher: Optional[bool] = None is_teacher_invitee: Optional[bool] = None