From cb309eb87c850a631e64d11291c335b189afe419 Mon Sep 17 00:00:00 2001 From: Niklas Raab Date: Thu, 1 Oct 2026 01:35:59 +0200 Subject: [PATCH 1/5] Support company and child accounts via --account A Trade Republic login can act for more accounts than its own, e.g. a company account (legal entity) or a child's account. These are separate customers: the web app lists them at /api/v1/customer/relationships/detailed and switches by posting the customer id as subjectId to /api/v2/auth/web/session, which swaps the session tokens. - add relationships() and switch_account() to the API, and --account to every subcommand that logs in - keep the session on the chosen account when it is refreshed - read the securities account number from the accountPairs topic when /api/v2/auth/account refuses with INVALID_AUTH_ACCOUNT_STATE, which it does while the session acts for another account - accept saved cookies that act for another account instead of forcing a new login, and return to the own account when --account is not given Refs #228 Co-Authored-By: Claude Fable 5.1 --- README.md | 15 ++ pytr/account.py | 22 ++- pytr/api.py | 124 +++++++++++++- pytr/main.py | 17 ++ tests/test_accounts.py | 369 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 541 insertions(+), 6 deletions(-) create mode 100644 tests/test_accounts.py diff --git a/README.md b/README.md index 8e16e438..abfcb0a2 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ __Table of Contents__ * [Usage](#usage) * [Authentication](#authentication) * [Web login](#web-login) + * [Company and child accounts](#company-and-child-accounts) * [Development](#development) * [Setting Up a Development Environment](#setting-up-a-development-environment) * [Linting and Code Formatting](#linting-and-code-formatting) @@ -116,6 +117,20 @@ Web login uses the public web-login endpoints at `api.traderepublic.com`. Two va Both variants keep you logged in on your primary device, but you may need to re-authenticate every so often when running `pytr`. +### Company and child accounts + +A Trade Republic login can hold more than its own account, e.g. a company account or a child's account. Trade +Republic treats each of them as a customer of its own, and the web app switches between them after the login. Pass +`--account` to do the same, on any subcommand that performs a login: + +```sh +pytr portfolio --account LEGAL_ENTITY +pytr dl_docs --account "Mustermann Holding GmbH" ./docs +``` + +The value is the account's type or its name. An unknown value makes pytr list the accounts your login has. Without +`--account`, pytr uses your own account, as before. + ## Development ### Setting Up a Development Environment diff --git a/pytr/account.py b/pytr/account.py index 07ab2e32..afa8a619 100644 --- a/pytr/account.py +++ b/pytr/account.py @@ -4,6 +4,7 @@ import time from getpass import getpass +import requests from pygments import formatters, highlight, lexers from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi @@ -19,10 +20,12 @@ def get_settings(tr): return formatted_json -def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False): +def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False, account=None): """ Handle credentials parameters and store to credentials file if requested. If no parameters are set but are needed then ask for input + + `account` selects another account of this login, e.g. a company or child account. """ log = get_logger(__name__) save_cookies = True @@ -88,5 +91,20 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh tr.complete_weblogin(code) log.info("Logged in.") - log.debug(get_settings(tr)) + try: + settings = get_settings(tr) + except requests.exceptions.HTTPError as e: + if not tr._acts_for_other_account(e.response): + raise + # Saved cookies can still act for the account chosen in an earlier run. + settings = None + + if account is not None or settings is None: + try: + tr.switch_account(account) + except ValueError as e: + log.fatal(str(e)) + sys.exit(1) + else: + log.debug(settings) return tr diff --git a/pytr/api.py b/pytr/api.py index 4822b0fd..289de3d6 100644 --- a/pytr/api.py +++ b/pytr/api.py @@ -87,6 +87,28 @@ WEB_PLATFORM = "web-pro" +def _relationship_names(relationship): + """Everything a relationship can be selected by, lower-cased.""" + full_name = " ".join(filter(None, [relationship.get("firstName"), relationship.get("lastName")])) + names = [ + relationship.get("customerId"), + relationship.get("relationshipType"), + relationship.get("accountType"), + relationship.get("accountName"), + relationship.get("shortDisplayName"), + full_name, + ] + return {name.strip().casefold() for name in names if isinstance(name, str) and name.strip()} + + +def _describe_relationship(relationship): + name = relationship.get("accountName") or " ".join( + filter(None, [relationship.get("firstName"), relationship.get("lastName")]) + ) + kind = relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN" + return f"{kind} ({name})" if name else kind + + class TradeRepublicApi: _default_headers = { "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36" @@ -148,6 +170,8 @@ def __init__( if self._save_cookies: self._websession.cookies = MozillaCookieJar(self._cookies_file) self._sec_acc_no: str | None = None + # Customer id of the account the session acts for, once it is not the login's own. + self._subject_id: str | None = None def _fetch_waf_token_awswaf(self): """ @@ -504,7 +528,11 @@ def resume_websession(self): try: self.log.debug("Calling settings...") self.settings() - except requests.exceptions.HTTPError: + except requests.exceptions.HTTPError as e: + if self._acts_for_other_account(e.response): + # The cookies are fine, they were saved while acting for another account. + self.log.info("Websession resumed.") + return True self.log.info("Resuming websession failed.") self.log.debug("Error calling tr.settings().", exc_info=True) # in case the websession can not be resumed, start with a fresh set of cookies @@ -516,11 +544,81 @@ def resume_websession(self): def _web_request(self, url_path, payload=None, method="GET"): if self._session_expires_at < time.time(): - r = self._websession.get(f"{self._host}/api/v1/auth/web/session") + if self._subject_id is None: + r = self._websession.get(f"{self._host}/api/v1/auth/web/session") + else: + # Refreshing without the subject would not keep the session on that account. + r = self._post_session(self._subject_id) r.raise_for_status() self._session_expires_at = time.time() + 290 return self._websession.request(method=method, url=f"{self._host}{url_path}", data=payload) + def _post_session(self, subject_id): + """Ask for session tokens that act for `subject_id`, the way the web frontend does.""" + return self._websession.post( + f"{self._host}/api/v2/auth/web/session", + json={"subjectId": subject_id}, + headers=self._login_headers(), + ) + + @staticmethod + def _acts_for_other_account(response): + """Tell whether `/api/v2/auth/account` refused because the session acts for another account.""" + if response is None or response.status_code != 400: + return False + try: + return response.json()["errors"][0]["errorCode"] == "INVALID_AUTH_ACCOUNT_STATE" + except (ValueError, KeyError, IndexError, TypeError): + return False + + def relationships(self): + """List the accounts this login can act for. + + Besides the login's own account (`relationshipType` `SELF`) these are e.g. a + company account (`LEGAL_ENTITY_ACTOR`) or a child's account. + """ + r = self._web_request("/api/v1/customer/relationships/detailed") + r.raise_for_status() + return r.json().get("relationships", []) + + def _find_relationship(self, account): + """Pick one relationship by customer id, type or name. `None` picks the login's own account.""" + relationships = self.relationships() + wanted = "self" if account is None else account.strip().casefold() + matches = [r for r in relationships if wanted in _relationship_names(r)] + if len(matches) == 1: + return matches[0] + known = ", ".join(_describe_relationship(r) for r in relationships) or "none" + problem = "matches more than one account" if matches else "does not match any account" + raise ValueError(f"{account!r} {problem}. Available accounts: {known}.") + + def switch_account(self, account=None): + """Let the session act for another account of this login, e.g. a company or child account. + + `account` is a customer id, a relationship or account type (`LEGAL_ENTITY`), or the + account's name, as listed by `relationships()`. `None` switches back to the login's + own account. Returns the chosen relationship. + + Trade Republic does not address these accounts by a second account number. It hands + out new session tokens that act for the other customer, so every call made afterwards, + the websocket included, sees that account only. + """ + if self._ws is not None and self._ws.close_code is None: + raise ValueError("Close the websocket connection before switching the account.") + + relationship = self._find_relationship(account) + r = self._post_session(relationship["customerId"]) + r.raise_for_status() + + is_self = relationship.get("relationshipType") == "SELF" + self._subject_id = None if is_self else relationship["customerId"] + self._session_expires_at = time.time() + 290 + # The securities account belongs to the previous account. + self._sec_acc_no = None + self.save_websession() + self.log.info(f"Using account: {_describe_relationship(relationship)}") + return relationship + async def _get_ws(self): if self._ws and self._ws.close_code is None: return self._ws @@ -668,12 +766,30 @@ async def portfolio_status(self): return await self.subscribe({"type": "portfolioStatus"}) async def compact_portfolio(self): + if self._sec_acc_no is None and self._subject_id is None: + try: + self.settings() + except requests.exceptions.HTTPError as e: + if not self._acts_for_other_account(e.response): + raise if self._sec_acc_no is None: - self.settings() + # The account settings are only served for the login's own account. + self._sec_acc_no = await self._sec_acc_no_from_account_pairs() if self._sec_acc_no is None: - raise ValueError("Could not retrieve securities account number from account settings.") + raise ValueError("Could not retrieve securities account number.") return await self.subscribe({"type": "compactPortfolioByType", "secAccNo": self._sec_acc_no}) + async def account_pairs(self): + return await self.subscribe({"type": "accountPairs"}) + + async def _sec_acc_no_from_account_pairs(self, timeout=5.0): + """Read the securities account number of the account the session acts for.""" + response = await self._receive_one(self.account_pairs(), timeout=timeout) + accounts = [a for a in response.get("accounts", []) if a.get("securitiesAccountNumber")] + # An account can hold more than one pair; the portfolio is the default product. + accounts.sort(key=lambda a: a.get("productType") != "DEFAULT") + return accounts[0]["securitiesAccountNumber"] if accounts else None + async def watchlist(self): return await self.subscribe({"type": "watchlist"}) diff --git a/pytr/main.py b/pytr/main.py index ecc4a87c..0bc1ef00 100644 --- a/pytr/main.py +++ b/pytr/main.py @@ -100,6 +100,14 @@ def formatter(prog): const=None, dest="waf_token", ) + parser_login_args.add_argument( + "--account", + help=( + "Use another account of this login, e.g. a company or child account. Give its type " + '(e.g. "LEGAL_ENTITY") or its name. Defaults to your own account.' + ), + default=None, + ) parser_login_args.add_argument( "--store_credentials", help="Store credentials (Phone number, pin, cookies) for next usage", @@ -560,6 +568,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ) elif args.command == "portfolio": Portfolio( @@ -569,6 +578,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.include_watchlist, instruments_to_ignore=re.split(r"[,;]", args.ignore) if args.ignore else [], @@ -590,6 +600,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), isins, output=args.output, @@ -606,6 +617,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.isin, ).get() @@ -619,6 +631,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.output, args.format, @@ -653,6 +666,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.outputdir, not_before, @@ -688,6 +702,7 @@ def main(): pin=args.pin, store_credentials=args.store_credentials, waf_token=args.waf_token, + account=args.account, ), args.input, args.outputfile, @@ -703,6 +718,7 @@ def main(): pin=args.pin, store_credentials=args.store_credentials, waf_token=args.waf_token, + account=args.account, ), args.input, args.inputfile, @@ -719,6 +735,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.outputfile, decimal_localization=args.decimal_localization, diff --git a/tests/test_accounts.py b/tests/test_accounts.py new file mode 100644 index 00000000..2ccd1f56 --- /dev/null +++ b/tests/test_accounts.py @@ -0,0 +1,369 @@ +"""Pin how pytr reaches accounts other than the login's own: company and child accounts.""" + +import asyncio +import json as jsonlib +from typing import Any + +import pytest +import requests + +import pytr.account as account_module +from pytr.api import TradeRepublicApi + +HOST = "https://api.traderepublic.com" +SESSION_V1 = f"{HOST}/api/v1/auth/web/session" +SESSION_V2 = f"{HOST}/api/v2/auth/web/session" +RELATIONSHIPS = f"{HOST}/api/v1/customer/relationships/detailed" +ACCOUNT = f"{HOST}/api/v2/auth/account" + +SELF = { + "customerId": "id-self", + "firstName": "Erika", + "lastName": "Mustermann", + "relationshipType": "SELF", + "accountType": "ADULT", +} +COMPANY = { + "customerId": "id-company", + "relationshipType": "LEGAL_ENTITY_ACTOR", + "accountType": "LEGAL_ENTITY", + "accountName": "Mustermann Holding GmbH", +} +CHILD_A = {"customerId": "id-a", "firstName": "Max", "relationshipType": "PARENT", "accountType": "CHILD"} +CHILD_B = {"customerId": "id-b", "firstName": "Mia", "relationshipType": "PARENT", "accountType": "CHILD"} + +# What /api/v2/auth/account answers while the session acts for another account. +OTHER_ACCOUNT = (400, {"errors": [{"errorCode": "INVALID_AUTH_ACCOUNT_STATE", "errorMessage": None, "meta": None}]}) + + +class _Response: + def __init__(self, payload: Any = None, status_code: int = 200): + self.status_code = status_code + self._payload = {} if payload is None else payload + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise requests.HTTPError(f"{self.status_code} Client Error", response=self) # type: ignore[arg-type] + + def json(self) -> Any: + return self._payload + + +class _Session: + """Records requests instead of sending them. Replies come from a queue.""" + + def __init__(self, replies: list[Any]): + self.calls: list[dict[str, Any]] = [] + self._replies = list(replies) + self.headers = dict(TradeRepublicApi._default_headers) + + def _record(self, method: str, url: str, json: Any = None, headers: Any = None) -> _Response: + self.calls.append({"method": method, "url": url, "json": json, "headers": headers or {}}) + reply = self._replies.pop(0) if self._replies else {} + status, payload = reply if isinstance(reply, tuple) else (200, reply) + return _Response(payload, status) + + def post(self, url, json=None, headers=None): + return self._record("POST", url, json, headers) + + def get(self, url, headers=None): + return self._record("GET", url, None, headers) + + def request(self, method, url, data=None): + return self._record(method, url, None, None) + + +class _Websocket: + """Answers subscriptions from a table of topic -> payload.""" + + close_code = None + + def __init__(self, answers: dict[str, Any]): + self.sent: list[str] = [] + self._answers = answers + self._inbox: asyncio.Queue[str] = asyncio.Queue() + + async def send(self, message: str) -> None: + self.sent.append(message) + command, subscription_id, *payload = message.split(" ", 2) + if command == "sub": + topic = jsonlib.loads(payload[0])["type"] + if topic in self._answers: + await self._inbox.put(f"{subscription_id} A {jsonlib.dumps(self._answers[topic])}") + + async def recv(self) -> str: + return await self._inbox.get() + + def subscriptions(self) -> list[dict[str, Any]]: + return [jsonlib.loads(m.split(" ", 2)[2]) for m in self.sent if m.startswith("sub ")] + + +def _api(replies, session_is_fresh=True): + tr = TradeRepublicApi(phone_no="+490000000000", pin="0000", waf_token=None, save_cookies=False) + tr._websession = _Session(replies) + if session_is_fresh: + tr._session_expires_at = float("inf") + return tr + + +def _with_websocket(tr, answers): + ws = _Websocket(answers) + + async def get_ws(): + return ws + + tr._get_ws = get_ws + return ws + + +def _calls(tr): + return [(c["method"], c["url"]) for c in tr._websession.calls] + + +# --- listing the accounts ------------------------------------------------------------ + + +def test_relationships_lists_the_accounts_of_the_login(): + tr = _api([{"relationships": [SELF, COMPANY]}]) + + assert tr.relationships() == [SELF, COMPANY] + assert _calls(tr) == [("GET", RELATIONSHIPS)] + + +# --- switching ----------------------------------------------------------------------- + + +def test_switch_account_asks_for_a_session_acting_for_the_other_customer(): + tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr._sec_acc_no = "own-depot" + + chosen = tr.switch_account("LEGAL_ENTITY") + + assert chosen == COMPANY + assert _calls(tr) == [("GET", RELATIONSHIPS), ("POST", SESSION_V2)] + assert tr._websession.calls[1]["json"] == {"subjectId": "id-company"} + # The securities account of the own account must not leak into the other one. + assert tr._sec_acc_no is None + + +def test_switch_account_sends_the_headers_the_v2_auth_endpoints_require(): + tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + + tr.switch_account("LEGAL_ENTITY") + + for header in ("X-TR-Device-Info", "X-TR-App-Version", "X-Tr-Platform"): + assert tr._websession.calls[1]["headers"].get(header) + + +@pytest.mark.parametrize( + "account", + ["id-company", "legal_entity", "LEGAL_ENTITY_ACTOR", "mustermann holding gmbh", " Mustermann Holding GmbH "], +) +def test_switch_account_finds_the_account_by_id_type_or_name(account): + tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + + assert tr.switch_account(account) == COMPANY + + +def test_switch_account_without_a_name_returns_to_the_own_account(): + tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr._subject_id = "id-company" + + assert tr.switch_account() == SELF + assert tr._websession.calls[1]["json"] == {"subjectId": "id-self"} + assert tr._subject_id is None + + +def test_switch_account_rejects_an_unknown_account_and_names_the_known_ones(): + tr = _api([{"relationships": [SELF, COMPANY]}]) + + with pytest.raises(ValueError, match="does not match any account.*LEGAL_ENTITY \\(Mustermann Holding GmbH\\)"): + tr.switch_account("nope") + + assert _calls(tr) == [("GET", RELATIONSHIPS)] + + +def test_switch_account_rejects_an_ambiguous_account(): + """Two children share their type, so the type alone does not say which one is meant.""" + tr = _api([{"relationships": [SELF, CHILD_A, CHILD_B]}, {"relationships": [SELF, CHILD_A, CHILD_B]}, {}]) + + with pytest.raises(ValueError, match="more than one account"): + tr.switch_account("CHILD") + + assert tr.switch_account("Mia") == CHILD_B + + +def test_switch_account_reports_a_refused_switch(): + tr = _api([{"relationships": [SELF, COMPANY]}, (403, {})]) + + with pytest.raises(requests.HTTPError): + tr.switch_account("LEGAL_ENTITY") + + assert tr._subject_id is None + + +def test_switch_account_refuses_while_the_websocket_is_connected(): + """An open websocket keeps answering for the account it was authenticated with.""" + tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr._ws = _Websocket({}) + + with pytest.raises(ValueError, match="websocket"): + tr.switch_account("LEGAL_ENTITY") + + assert tr._websession.calls == [] + + +# --- keeping the session on the chosen account --------------------------------------- + + +def test_session_refresh_stays_on_the_chosen_account(): + tr = _api([{"relationships": [SELF, COMPANY]}, {}, {}, {}]) + tr.switch_account("LEGAL_ENTITY") + tr._session_expires_at = 0 + + tr._web_request("/api/v1/whatever") + + assert _calls(tr)[2] == ("POST", SESSION_V2) + assert tr._websession.calls[2]["json"] == {"subjectId": "id-company"} + + +def test_session_refresh_is_unchanged_for_the_own_account(): + tr = _api([{}, {}], session_is_fresh=False) + + tr._web_request("/api/v1/whatever") + + assert _calls(tr)[0] == ("GET", SESSION_V1) + + +def _resumable(tmp_path, replies): + cookies = tmp_path / "cookies.txt" + cookies.write_text("# Netscape HTTP Cookie File\n") + tr = TradeRepublicApi( + phone_no="+490000000000", pin="0000", waf_token=None, save_cookies=True, cookies_file=str(cookies) + ) + jar = tr._websession.cookies + tr._websession = _Session(replies) + tr._websession.cookies = jar # type: ignore[attr-defined] + return tr + + +def test_resume_accepts_cookies_saved_while_acting_for_another_account(tmp_path): + """Those cookies are valid; only the account settings are not served for them.""" + tr = _resumable(tmp_path, [{}, OTHER_ACCOUNT]) + + assert tr.resume_websession() is True + + +def test_resume_still_rejects_an_expired_session(tmp_path): + tr = _resumable(tmp_path, [(401, {})]) + + assert tr.resume_websession() is False + + +# --- portfolio ----------------------------------------------------------------------- + +PAIRS = { + "authAccountId": "id-company", + "accounts": [ + {"securitiesAccountNumber": "depot-other", "cashAccountNumber": "cash-other", "productType": "OTHER"}, + {"securitiesAccountNumber": "depot-company", "cashAccountNumber": "cash-company", "productType": "DEFAULT"}, + ], +} + + +def test_compact_portfolio_reads_the_securities_account_from_account_pairs_after_a_switch(): + tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr.switch_account("LEGAL_ENTITY") + ws = _with_websocket(tr, {"accountPairs": PAIRS}) + + asyncio.run(tr.compact_portfolio()) + + assert ws.subscriptions() == [ + {"type": "accountPairs"}, + {"type": "compactPortfolioByType", "secAccNo": "depot-company"}, + ] + # The account settings are not served for this account, so they are not asked for. + assert ("GET", ACCOUNT) not in _calls(tr) + + +def test_compact_portfolio_falls_back_to_account_pairs_when_the_settings_are_refused(): + tr = _api([OTHER_ACCOUNT]) + ws = _with_websocket(tr, {"accountPairs": PAIRS}) + + asyncio.run(tr.compact_portfolio()) + + assert ws.subscriptions()[-1] == {"type": "compactPortfolioByType", "secAccNo": "depot-company"} + + +def test_compact_portfolio_still_uses_the_account_settings_for_the_own_account(): + tr = _api([{"securitiesAccountNumber": "own-depot"}]) + ws = _with_websocket(tr, {}) + + asyncio.run(tr.compact_portfolio()) + + assert ws.subscriptions() == [{"type": "compactPortfolioByType", "secAccNo": "own-depot"}] + + +def test_compact_portfolio_does_not_hide_other_errors_of_the_settings_call(): + tr = _api([(500, {})]) + _with_websocket(tr, {"accountPairs": PAIRS}) + + with pytest.raises(requests.HTTPError): + asyncio.run(tr.compact_portfolio()) + + +def test_compact_portfolio_reports_a_missing_securities_account(): + tr = _api([{}]) + _with_websocket(tr, {"accountPairs": {"authAccountId": "id-self", "accounts": []}}) + + with pytest.raises(ValueError, match="securities account number"): + asyncio.run(tr.compact_portfolio()) + + +# --- login() ------------------------------------------------------------------------- + + +class _LoggedIn: + """Stands in for a TradeRepublicApi whose saved session could be resumed.""" + + def __init__(self, settings_reply): + self._settings_reply = settings_reply + self.switched_to: list[Any] = [] + + def resume_websession(self): + return True + + def settings(self): + status, payload = self._settings_reply + response = _Response(payload, status) + response.raise_for_status() + return payload + + _acts_for_other_account = staticmethod(TradeRepublicApi._acts_for_other_account) + + def switch_account(self, account=None): + self.switched_to.append(account) + + +def _login(monkeypatch, settings_reply, **kwargs): + tr = _LoggedIn(settings_reply) + monkeypatch.setattr(account_module, "TradeRepublicApi", lambda **_: tr) + assert account_module.login(phone_no="+490000000000", pin="0000", **kwargs) is tr + return tr + + +def test_login_does_not_switch_by_default(monkeypatch): + assert _login(monkeypatch, (200, {})).switched_to == [] + + +def test_login_switches_to_the_requested_account(monkeypatch): + assert _login(monkeypatch, (200, {}), account="LEGAL_ENTITY").switched_to == ["LEGAL_ENTITY"] + + +def test_login_stays_on_the_requested_account_when_the_cookies_already_act_for_it(monkeypatch): + assert _login(monkeypatch, OTHER_ACCOUNT, account="LEGAL_ENTITY").switched_to == ["LEGAL_ENTITY"] + + +def test_login_returns_to_the_own_account_when_none_is_requested(monkeypatch): + """Cookies saved by a run with --account must not make the next plain run use that account.""" + assert _login(monkeypatch, OTHER_ACCOUNT).switched_to == [None] From 20aed78347aefd85af2c1e8c2f4db44b60e74f0f Mon Sep 17 00:00:00 2001 From: Niklas Raab Date: Thu, 1 Oct 2026 01:38:16 +0200 Subject: [PATCH 2/5] Add an accounts command that lists what --account accepts Prints type, name and state of every account the login can act for, from the same relationships call the switch resolves its argument with. Refs #228 Co-Authored-By: Claude Fable 5.1 --- README.md | 10 ++++++---- pytr/account.py | 16 +++++++++++++++- pytr/api.py | 14 +++++++++++--- pytr/main.py | 25 ++++++++++++++++++++++++- tests/test_accounts.py | 25 +++++++++++++++++++++++++ 5 files changed, 81 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index abfcb0a2..b25a7119 100644 --- a/README.md +++ b/README.md @@ -62,16 +62,18 @@ If you want to use the cutting-edge version, use this command instead: ```console usage: pytr [-h] [-V] [-v {warning,info,debug}] [--debug-logfile DEBUG_LOGFILE] [--debug-log-filter DEBUG_LOG_FILTER] - {help,login,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ... + {help,login,accounts,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ... Use "pytr command_name --help" to get detailed help to a specific command Commands: - {help,login,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} + {help,login,accounts,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} Desired action to perform help Print this help message login Check if credentials file exists. If not create it and ask for input. Try to login. Ask for device reset if needed + accounts List the accounts of this login, e.g. company or child accounts. Use one of them + with --account portfolio Show current portfolio rates Fetch current prices for a list of ISINs given as direct list or CSV input details Get details for an ISIN @@ -128,8 +130,8 @@ pytr portfolio --account LEGAL_ENTITY pytr dl_docs --account "Mustermann Holding GmbH" ./docs ``` -The value is the account's type or its name. An unknown value makes pytr list the accounts your login has. Without -`--account`, pytr uses your own account, as before. +The value is the account's type or its name, as listed by `pytr accounts`. Without `--account`, pytr uses your own +account, as before. ## Development diff --git a/pytr/account.py b/pytr/account.py index afa8a619..e4ff466f 100644 --- a/pytr/account.py +++ b/pytr/account.py @@ -7,7 +7,7 @@ import requests from pygments import formatters, highlight, lexers -from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi +from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi, relationship_name, relationship_type from .utils import get_logger @@ -20,6 +20,20 @@ def get_settings(tr): return formatted_json +def get_accounts(tr): + """List the accounts of this login, one per line: type, name and state.""" + rows = [ + ( + relationship_type(r), + relationship_name(r), + "your own account" if r.get("relationshipType") == "SELF" else (r.get("accountState") or ""), + ) + for r in tr.relationships() + ] + widths = [max((len(row[i]) for row in rows), default=0) for i in range(2)] + return "\n".join(f"{kind:<{widths[0]}} {name:<{widths[1]}} {note}".rstrip() for kind, name, note in rows) + + def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False, account=None): """ Handle credentials parameters and store to credentials file if requested. diff --git a/pytr/api.py b/pytr/api.py index 289de3d6..899b091b 100644 --- a/pytr/api.py +++ b/pytr/api.py @@ -101,11 +101,19 @@ def _relationship_names(relationship): return {name.strip().casefold() for name in names if isinstance(name, str) and name.strip()} -def _describe_relationship(relationship): - name = relationship.get("accountName") or " ".join( +def relationship_type(relationship): + return relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN" + + +def relationship_name(relationship): + return relationship.get("accountName") or " ".join( filter(None, [relationship.get("firstName"), relationship.get("lastName")]) ) - kind = relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN" + + +def _describe_relationship(relationship): + name = relationship_name(relationship) + kind = relationship_type(relationship) return f"{kind} ({name})" if name else kind diff --git a/pytr/main.py b/pytr/main.py index 0bc1ef00..3b255271 100644 --- a/pytr/main.py +++ b/pytr/main.py @@ -12,7 +12,7 @@ import shtab -from pytr.account import login +from pytr.account import get_accounts, login from pytr.alarms import Alarms from pytr.details import Details from pytr.dl import DL @@ -174,6 +174,16 @@ def formatter(prog): description=info, ) + # accounts + info = "List the accounts of this login, e.g. company or child accounts. Use one of them with --account" + parser_cmd.add_parser( + "accounts", + formatter_class=formatter, + parents=[parser_login_args], + help=info, + description=info, + ) + # portfolio info = "Show current portfolio" parser_portfolio = parser_cmd.add_parser( @@ -570,6 +580,19 @@ def main(): v2=args.v2, account=args.account, ) + elif args.command == "accounts": + print( + get_accounts( + login( + phone_no=args.phone_no, + pin=args.pin, + store_credentials=args.store_credentials, + waf_token=args.waf_token, + v2=args.v2, + account=args.account, + ) + ) + ) elif args.command == "portfolio": Portfolio( login( diff --git a/tests/test_accounts.py b/tests/test_accounts.py index 2ccd1f56..6eb151bc 100644 --- a/tests/test_accounts.py +++ b/tests/test_accounts.py @@ -320,6 +320,31 @@ def test_compact_portfolio_reports_a_missing_securities_account(): asyncio.run(tr.compact_portfolio()) +# --- pytr accounts ------------------------------------------------------------------- + + +def test_get_accounts_prints_what_account_accepts(): + tr = _api([{"relationships": [SELF, {**COMPANY, "accountState": "ACTIVE"}, CHILD_B]}]) + + assert account_module.get_accounts(tr).splitlines() == [ + "ADULT Erika Mustermann your own account", + "LEGAL_ENTITY Mustermann Holding GmbH ACTIVE", + "CHILD Mia", + ] + + +def test_get_accounts_of_a_login_without_relationships_is_empty(): + assert account_module.get_accounts(_api([{}])) == "" + + +def test_the_accounts_command_exists_and_takes_the_login_arguments(): + from pytr.main import get_main_parser + + args = get_main_parser().parse_args(["accounts", "--v2"]) + + assert (args.command, args.v2, args.account) == ("accounts", True, None) + + # --- login() ------------------------------------------------------------------------- From 4ba9f7313f1088162cee24a9f0c7444ee8c28d14 Mon Sep 17 00:00:00 2001 From: Niklas Raab Date: Thu, 1 Oct 2026 02:02:16 +0200 Subject: [PATCH 3/5] Simplify account switching after live tests and review Live tests showed that the session refresh without a subject returns the session to the login's own account, and that a new process always starts with that refresh. So saved cookies never carry the chosen account into the next run, and the handling added for that case could not be reached. - drop the INVALID_AUTH_ACCOUNT_STATE handling in resume_websession(), login() and compact_portfolio(); the own-account paths are as before - keep answers to other subscriptions while compact_portfolio() looks up the securities account, instead of discarding them - report a missing or unanswered accountPairs lookup as ValueError - exit with a message when the switch is refused, not a traceback - match accounts by customer id, type, name or first name; show the customer id in 'pytr accounts' for accounts that share type and name - tolerate missing or non-text fields in the relationships list - stop claiming child accounts work: only a company account was tested - tests: fail on unexpected requests, isolate login() from ~/.pytr, pin the request order of a new process, cover the error paths Refs #228 Co-Authored-By: Claude Fable 5.1 --- README.md | 21 +-- pytr/account.py | 36 ++--- pytr/api.py | 155 ++++++++++-------- pytr/main.py | 7 +- tests/test_accounts.py | 346 +++++++++++++++++++++++++++-------------- 5 files changed, 352 insertions(+), 213 deletions(-) diff --git a/README.md b/README.md index b25a7119..448150cc 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ __Table of Contents__ * [Usage](#usage) * [Authentication](#authentication) * [Web login](#web-login) - * [Company and child accounts](#company-and-child-accounts) + * [Other accounts of your login](#other-accounts-of-your-login) * [Development](#development) * [Setting Up a Development Environment](#setting-up-a-development-environment) * [Linting and Code Formatting](#linting-and-code-formatting) @@ -72,8 +72,8 @@ Commands: help Print this help message login Check if credentials file exists. If not create it and ask for input. Try to login. Ask for device reset if needed - accounts List the accounts of this login, e.g. company or child accounts. Use one of them - with --account + accounts List the accounts of this login, e.g. a company account. Use one of them with + --account portfolio Show current portfolio rates Fetch current prices for a list of ISINs given as direct list or CSV input details Get details for an ISIN @@ -119,19 +119,22 @@ Web login uses the public web-login endpoints at `api.traderepublic.com`. Two va Both variants keep you logged in on your primary device, but you may need to re-authenticate every so often when running `pytr`. -### Company and child accounts +### Other accounts of your login -A Trade Republic login can hold more than its own account, e.g. a company account or a child's account. Trade -Republic treats each of them as a customer of its own, and the web app switches between them after the login. Pass -`--account` to do the same, on any subcommand that performs a login: +A Trade Republic login can hold more than its own account, e.g. a company account. Trade Republic treats each of +them as a customer of its own, and the web app switches between them after the login. Pass `--account` to do the +same, on any subcommand that performs a login: ```sh pytr portfolio --account LEGAL_ENTITY pytr dl_docs --account "Mustermann Holding GmbH" ./docs ``` -The value is the account's type or its name, as listed by `pytr accounts`. Without `--account`, pytr uses your own -account, as before. +The value is the account's type, its name or its customer id, as listed by `pytr accounts`. Without `--account`, +pytr uses your own account, as before. The choice is not remembered between runs. + +This has been tested with a company account. Other accounts that the web app lets you switch to, e.g. a child's +account, are expected to work the same way but have not been tested. ## Development diff --git a/pytr/account.py b/pytr/account.py index e4ff466f..1c322695 100644 --- a/pytr/account.py +++ b/pytr/account.py @@ -7,7 +7,7 @@ import requests from pygments import formatters, highlight, lexers -from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi, relationship_name, relationship_type +from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi, account_kind, account_name from .utils import get_logger @@ -21,17 +21,21 @@ def get_settings(tr): def get_accounts(tr): - """List the accounts of this login, one per line: type, name and state.""" + """List the accounts of this login, one per line: type, name, customer id and state. + + `--account` takes the type, the name or, where those are not unique, the customer id. + """ rows = [ ( - relationship_type(r), - relationship_name(r), - "your own account" if r.get("relationshipType") == "SELF" else (r.get("accountState") or ""), + account_kind(r), + account_name(r), + str(r.get("customerId") or ""), + "your own account" if r.get("relationshipType") == "SELF" else str(r.get("accountState") or ""), ) for r in tr.relationships() ] - widths = [max((len(row[i]) for row in rows), default=0) for i in range(2)] - return "\n".join(f"{kind:<{widths[0]}} {name:<{widths[1]}} {note}".rstrip() for kind, name, note in rows) + widths = [max((len(row[i]) for row in rows), default=0) for i in range(3)] + return "\n".join(" ".join(f"{cell:<{width}}" for cell, width in zip(row, widths + [0])).rstrip() for row in rows) def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False, account=None): @@ -39,7 +43,7 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh Handle credentials parameters and store to credentials file if requested. If no parameters are set but are needed then ask for input - `account` selects another account of this login, e.g. a company or child account. + `account` selects another account of this login, e.g. a company account. """ log = get_logger(__name__) save_cookies = True @@ -105,20 +109,16 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh tr.complete_weblogin(code) log.info("Logged in.") - try: - settings = get_settings(tr) - except requests.exceptions.HTTPError as e: - if not tr._acts_for_other_account(e.response): - raise - # Saved cookies can still act for the account chosen in an earlier run. - settings = None + log.debug(get_settings(tr)) - if account is not None or settings is None: + if account is not None: try: tr.switch_account(account) except ValueError as e: log.fatal(str(e)) sys.exit(1) - else: - log.debug(settings) + except requests.exceptions.HTTPError as e: + status = e.response.status_code if e.response is not None else "unknown" + log.fatal(f"Trade Republic refused to switch to account {account!r} (status {status}).") + sys.exit(1) return tr diff --git a/pytr/api.py b/pytr/api.py index 899b091b..c6bf1ee7 100644 --- a/pytr/api.py +++ b/pytr/api.py @@ -32,6 +32,7 @@ import time import urllib.parse import uuid +from collections import deque from datetime import datetime from http.cookiejar import Cookie, MozillaCookieJar from typing import Any, Dict @@ -87,34 +88,36 @@ WEB_PLATFORM = "web-pro" -def _relationship_names(relationship): - """Everything a relationship can be selected by, lower-cased.""" - full_name = " ".join(filter(None, [relationship.get("firstName"), relationship.get("lastName")])) - names = [ - relationship.get("customerId"), - relationship.get("relationshipType"), - relationship.get("accountType"), - relationship.get("accountName"), - relationship.get("shortDisplayName"), - full_name, - ] - return {name.strip().casefold() for name in names if isinstance(name, str) and name.strip()} +# Seconds after which the web session is refreshed. +SESSION_REFRESH_INTERVAL = 290 -def relationship_type(relationship): - return relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN" +def account_kind(relationship): + """The kind of account behind an entry of `relationships()`, e.g. `ADULT` or `LEGAL_ENTITY`.""" + return str(relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN") -def relationship_name(relationship): - return relationship.get("accountName") or " ".join( - filter(None, [relationship.get("firstName"), relationship.get("lastName")]) - ) +def account_name(relationship): + """The name of the account behind an entry of `relationships()`: the account's, else the person's.""" + person = [str(part) for part in (relationship.get("firstName"), relationship.get("lastName")) if part] + return str(relationship.get("accountName") or " ".join(person)) -def _describe_relationship(relationship): - name = relationship_name(relationship) - kind = relationship_type(relationship) - return f"{kind} ({name})" if name else kind +def _account_selectors(relationship): + """Everything an entry of `relationships()` can be selected by, lower-cased.""" + selectors = [ + relationship.get("customerId"), + relationship.get("relationshipType"), + relationship.get("accountType"), + relationship.get("firstName"), + account_name(relationship), + ] + return {str(selector).strip().casefold() for selector in selectors if selector and str(selector).strip()} + + +def _describe_account(relationship): + name = account_name(relationship) + return f"{account_kind(relationship)} ({name})" if name else account_kind(relationship) class TradeRepublicApi: @@ -180,6 +183,8 @@ def __init__( self._sec_acc_no: str | None = None # Customer id of the account the session acts for, once it is not the login's own. self._subject_id: str | None = None + # Messages read off the websocket on behalf of a caller that has not asked for them yet. + self._held_back: deque = deque() def _fetch_waf_token_awswaf(self): """ @@ -536,11 +541,7 @@ def resume_websession(self): try: self.log.debug("Calling settings...") self.settings() - except requests.exceptions.HTTPError as e: - if self._acts_for_other_account(e.response): - # The cookies are fine, they were saved while acting for another account. - self.log.info("Websession resumed.") - return True + except requests.exceptions.HTTPError: self.log.info("Resuming websession failed.") self.log.debug("Error calling tr.settings().", exc_info=True) # in case the websession can not be resumed, start with a fresh set of cookies @@ -553,62 +554,61 @@ def resume_websession(self): def _web_request(self, url_path, payload=None, method="GET"): if self._session_expires_at < time.time(): if self._subject_id is None: + # Refreshing without a subject returns the session to the login's own account. + # A new process starts here, so it never inherits the account an earlier run chose. r = self._websession.get(f"{self._host}/api/v1/auth/web/session") else: - # Refreshing without the subject would not keep the session on that account. r = self._post_session(self._subject_id) r.raise_for_status() - self._session_expires_at = time.time() + 290 + self._session_expires_at = time.time() + SESSION_REFRESH_INTERVAL return self._websession.request(method=method, url=f"{self._host}{url_path}", data=payload) def _post_session(self, subject_id): - """Ask for session tokens that act for `subject_id`, the way the web frontend does.""" + """Ask for session cookies that act for `subject_id`, the way the web frontend does. + + The frontend uses this call both to switch the account and to refresh the session. + """ return self._websession.post( f"{self._host}/api/v2/auth/web/session", json={"subjectId": subject_id}, headers=self._login_headers(), ) - @staticmethod - def _acts_for_other_account(response): - """Tell whether `/api/v2/auth/account` refused because the session acts for another account.""" - if response is None or response.status_code != 400: - return False - try: - return response.json()["errors"][0]["errorCode"] == "INVALID_AUTH_ACCOUNT_STATE" - except (ValueError, KeyError, IndexError, TypeError): - return False - def relationships(self): """List the accounts this login can act for. - Besides the login's own account (`relationshipType` `SELF`) these are e.g. a - company account (`LEGAL_ENTITY_ACTOR`) or a child's account. + Besides the login's own account (`relationshipType` `SELF`) this is e.g. a + company account (`LEGAL_ENTITY_ACTOR`). """ r = self._web_request("/api/v1/customer/relationships/detailed") r.raise_for_status() - return r.json().get("relationships", []) + return r.json().get("relationships") or [] def _find_relationship(self, account): """Pick one relationship by customer id, type or name. `None` picks the login's own account.""" relationships = self.relationships() - wanted = "self" if account is None else account.strip().casefold() - matches = [r for r in relationships if wanted in _relationship_names(r)] - if len(matches) == 1: + if account is None: + matches = [r for r in relationships if r.get("relationshipType") == "SELF"] + wanted = "Your own account" + else: + matches = [r for r in relationships if account.strip().casefold() in _account_selectors(r)] + wanted = repr(account) + if len(matches) == 1 and matches[0].get("customerId"): return matches[0] - known = ", ".join(_describe_relationship(r) for r in relationships) or "none" - problem = "matches more than one account" if matches else "does not match any account" - raise ValueError(f"{account!r} {problem}. Available accounts: {known}.") + known = ", ".join(_describe_account(r) for r in relationships) or "none" + if len(matches) > 1: + raise ValueError(f"{wanted} matches more than one account, use its customer id. Accounts: {known}.") + raise ValueError(f"{wanted} was not found among the accounts of this login. Accounts: {known}.") def switch_account(self, account=None): - """Let the session act for another account of this login, e.g. a company or child account. + """Let the session act for another account of this login, e.g. a company account. - `account` is a customer id, a relationship or account type (`LEGAL_ENTITY`), or the - account's name, as listed by `relationships()`. `None` switches back to the login's - own account. Returns the chosen relationship. + `account` is the account's type (`LEGAL_ENTITY`), its name or its customer id, as listed + by `relationships()`. `None` switches back to the login's own account. Returns the chosen + relationship. Trade Republic does not address these accounts by a second account number. It hands - out new session tokens that act for the other customer, so every call made afterwards, + out new session cookies that act for the other customer, so every call made afterwards, the websocket included, sees that account only. """ if self._ws is not None and self._ws.close_code is None: @@ -620,11 +620,12 @@ def switch_account(self, account=None): is_self = relationship.get("relationshipType") == "SELF" self._subject_id = None if is_self else relationship["customerId"] - self._session_expires_at = time.time() + 290 + self._session_expires_at = time.time() + SESSION_REFRESH_INTERVAL # The securities account belongs to the previous account. self._sec_acc_no = None + # The switch replaced the session cookies; keep the saved ones usable for the next run. self.save_websession() - self.log.info(f"Using account: {_describe_relationship(relationship)}") + self.log.info(f"Using account: {_describe_account(relationship)}") return relationship async def _get_ws(self): @@ -697,6 +698,11 @@ async def unsubscribe(self, subscription_id): self._previous_responses.pop(subscription_id, None) async def recv(self): + if self._held_back: + return self._held_back.popleft() + return await self._recv_ws() + + async def _recv_ws(self): ws = await self._get_ws() while True: response = await ws.recv() @@ -774,17 +780,16 @@ async def portfolio_status(self): return await self.subscribe({"type": "portfolioStatus"}) async def compact_portfolio(self): - if self._sec_acc_no is None and self._subject_id is None: - try: - self.settings() - except requests.exceptions.HTTPError as e: - if not self._acts_for_other_account(e.response): - raise if self._sec_acc_no is None: - # The account settings are only served for the login's own account. - self._sec_acc_no = await self._sec_acc_no_from_account_pairs() + if self._subject_id is None: + self.settings() + else: + # /api/v2/auth/account answers 400 INVALID_AUTH_ACCOUNT_STATE for any account but + # the login's own, so the number has to come from the websocket instead. + self._sec_acc_no = await self._sec_acc_no_from_account_pairs() if self._sec_acc_no is None: - raise ValueError("Could not retrieve securities account number.") + source = "account settings" if self._subject_id is None else "account pairs" + raise ValueError(f"Could not retrieve securities account number from {source}.") return await self.subscribe({"type": "compactPortfolioByType", "secAccNo": self._sec_acc_no}) async def account_pairs(self): @@ -792,12 +797,26 @@ async def account_pairs(self): async def _sec_acc_no_from_account_pairs(self, timeout=5.0): """Read the securities account number of the account the session acts for.""" - response = await self._receive_one(self.account_pairs(), timeout=timeout) - accounts = [a for a in response.get("accounts", []) if a.get("securitiesAccountNumber")] - # An account can hold more than one pair; the portfolio is the default product. + subscription_id = await self.account_pairs() + try: + response = await asyncio.wait_for(self._recv_holding_back_others(subscription_id), timeout) + except asyncio.TimeoutError: + return None + finally: + await self.unsubscribe(subscription_id) + accounts = [a for a in response.get("accounts") or [] if a.get("securitiesAccountNumber")] + # Prefer the pair of the default product if the account has more than one. accounts.sort(key=lambda a: a.get("productType") != "DEFAULT") return accounts[0]["securitiesAccountNumber"] if accounts else None + async def _recv_holding_back_others(self, subscription_id): + """Wait for one subscription's answer; keep the answers to others for the next `recv()`.""" + while True: + message = await self._recv_ws() + if message[0] == subscription_id: + return message[2] + self._held_back.append(message) + async def watchlist(self): return await self.subscribe({"type": "watchlist"}) diff --git a/pytr/main.py b/pytr/main.py index 3b255271..d13a5f7b 100644 --- a/pytr/main.py +++ b/pytr/main.py @@ -103,8 +103,9 @@ def formatter(prog): parser_login_args.add_argument( "--account", help=( - "Use another account of this login, e.g. a company or child account. Give its type " - '(e.g. "LEGAL_ENTITY") or its name. Defaults to your own account.' + "Use another account of this login, e.g. a company account. Give its type " + '(e.g. "LEGAL_ENTITY"), name or customer id as listed by "pytr accounts". ' + "Defaults to your own account." ), default=None, ) @@ -175,7 +176,7 @@ def formatter(prog): ) # accounts - info = "List the accounts of this login, e.g. company or child accounts. Use one of them with --account" + info = "List the accounts of this login, e.g. a company account. Use one of them with --account" parser_cmd.add_parser( "accounts", formatter_class=formatter, diff --git a/tests/test_accounts.py b/tests/test_accounts.py index 6eb151bc..2ca791ba 100644 --- a/tests/test_accounts.py +++ b/tests/test_accounts.py @@ -1,4 +1,4 @@ -"""Pin how pytr reaches accounts other than the login's own: company and child accounts.""" +"""Pin how pytr reaches an account of the login other than its own, e.g. a company account.""" import asyncio import json as jsonlib @@ -9,6 +9,7 @@ import pytr.account as account_module from pytr.api import TradeRepublicApi +from pytr.main import get_main_parser HOST = "https://api.traderepublic.com" SESSION_V1 = f"{HOST}/api/v1/auth/web/session" @@ -21,19 +22,18 @@ "firstName": "Erika", "lastName": "Mustermann", "relationshipType": "SELF", + "accountState": "ACTIVE", "accountType": "ADULT", } COMPANY = { "customerId": "id-company", "relationshipType": "LEGAL_ENTITY_ACTOR", + "accountState": "ACTIVE", "accountType": "LEGAL_ENTITY", "accountName": "Mustermann Holding GmbH", } -CHILD_A = {"customerId": "id-a", "firstName": "Max", "relationshipType": "PARENT", "accountType": "CHILD"} -CHILD_B = {"customerId": "id-b", "firstName": "Mia", "relationshipType": "PARENT", "accountType": "CHILD"} - -# What /api/v2/auth/account answers while the session acts for another account. -OTHER_ACCOUNT = (400, {"errors": [{"errorCode": "INVALID_AUTH_ACCOUNT_STATE", "errorMessage": None, "meta": None}]}) +SECOND_COMPANY = {**COMPANY, "customerId": "id-second", "accountName": "Mustermann Immobilien UG"} +BOTH = {"relationships": [SELF, COMPANY]} class _Response: @@ -41,6 +41,10 @@ def __init__(self, payload: Any = None, status_code: int = 200): self.status_code = status_code self._payload = {} if payload is None else payload + def __bool__(self) -> bool: + # Like requests.Response: falsy for 4xx and 5xx. + return self.status_code < 400 + def raise_for_status(self) -> None: if self.status_code >= 400: raise requests.HTTPError(f"{self.status_code} Client Error", response=self) # type: ignore[arg-type] @@ -50,16 +54,17 @@ def json(self) -> Any: class _Session: - """Records requests instead of sending them. Replies come from a queue.""" + """Records requests instead of sending them. Replies come from a queue; an unexpected request fails.""" def __init__(self, replies: list[Any]): self.calls: list[dict[str, Any]] = [] - self._replies = list(replies) + self.replies = list(replies) self.headers = dict(TradeRepublicApi._default_headers) def _record(self, method: str, url: str, json: Any = None, headers: Any = None) -> _Response: self.calls.append({"method": method, "url": url, "json": json, "headers": headers or {}}) - reply = self._replies.pop(0) if self._replies else {} + assert self.replies, f"unexpected request: {method} {url}" + reply = self.replies.pop(0) status, payload = reply if isinstance(reply, tuple) else (200, reply) return _Response(payload, status) @@ -70,26 +75,29 @@ def get(self, url, headers=None): return self._record("GET", url, None, headers) def request(self, method, url, data=None): - return self._record(method, url, None, None) + return self._record(method, url, data, None) class _Websocket: - """Answers subscriptions from a table of topic -> payload.""" + """Answers each subscription with the messages listed for its topic: (code, topic answered, payload).""" close_code = None - def __init__(self, answers: dict[str, Any]): + def __init__(self, answers: dict[str, list[tuple[str, str, Any]]]): self.sent: list[str] = [] self._answers = answers + self._ids: dict[str, str] = {} self._inbox: asyncio.Queue[str] = asyncio.Queue() async def send(self, message: str) -> None: self.sent.append(message) command, subscription_id, *payload = message.split(" ", 2) - if command == "sub": - topic = jsonlib.loads(payload[0])["type"] - if topic in self._answers: - await self._inbox.put(f"{subscription_id} A {jsonlib.dumps(self._answers[topic])}") + if command != "sub": + return + topic = jsonlib.loads(payload[0])["type"] + self._ids[topic] = subscription_id + for code, answered_topic, body in self._answers.get(topic, []): + await self._inbox.put(f"{self._ids[answered_topic]} {code} {jsonlib.dumps(body)}") async def recv(self) -> str: return await self._inbox.get() @@ -101,11 +109,23 @@ def subscriptions(self) -> list[dict[str, Any]]: def _api(replies, session_is_fresh=True): tr = TradeRepublicApi(phone_no="+490000000000", pin="0000", waf_token=None, save_cookies=False) tr._websession = _Session(replies) + # These are class attributes; keep what one test subscribes to out of the next. + tr.subscriptions = {} + tr._previous_responses = {} + tr._subscription_id_counter = 1 if session_is_fresh: tr._session_expires_at = float("inf") return tr +def _switched(extra_replies=()): + """An API whose session acts for the company account.""" + tr = _api([BOTH, {}, *extra_replies]) + tr.switch_account("LEGAL_ENTITY") + tr._websession.calls.clear() + return tr + + def _with_websocket(tr, answers): ws = _Websocket(answers) @@ -124,17 +144,22 @@ def _calls(tr): def test_relationships_lists_the_accounts_of_the_login(): - tr = _api([{"relationships": [SELF, COMPANY]}]) + tr = _api([BOTH]) assert tr.relationships() == [SELF, COMPANY] assert _calls(tr) == [("GET", RELATIONSHIPS)] +@pytest.mark.parametrize("payload", [{}, {"relationships": None}]) +def test_relationships_of_a_login_without_any_is_empty(payload): + assert _api([payload]).relationships() == [] + + # --- switching ----------------------------------------------------------------------- def test_switch_account_asks_for_a_session_acting_for_the_other_customer(): - tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr = _api([BOTH, {}]) tr._sec_acc_no = "own-depot" chosen = tr.switch_account("LEGAL_ENTITY") @@ -142,12 +167,13 @@ def test_switch_account_asks_for_a_session_acting_for_the_other_customer(): assert chosen == COMPANY assert _calls(tr) == [("GET", RELATIONSHIPS), ("POST", SESSION_V2)] assert tr._websession.calls[1]["json"] == {"subjectId": "id-company"} + assert tr._subject_id == "id-company" # The securities account of the own account must not leak into the other one. assert tr._sec_acc_no is None -def test_switch_account_sends_the_headers_the_v2_auth_endpoints_require(): - tr = _api([{"relationships": [SELF, COMPANY]}, {}]) +def test_switch_account_sends_the_headers_of_the_v2_login_calls(): + tr = _api([BOTH, {}]) tr.switch_account("LEGAL_ENTITY") @@ -155,56 +181,105 @@ def test_switch_account_sends_the_headers_the_v2_auth_endpoints_require(): assert tr._websession.calls[1]["headers"].get(header) +def test_switch_account_saves_the_new_cookies(): + tr = _api([BOTH, {}]) + saved = [] + tr.save_websession = lambda: saved.append(tr._subject_id) + + tr.switch_account("LEGAL_ENTITY") + + assert saved == ["id-company"] + + @pytest.mark.parametrize( "account", ["id-company", "legal_entity", "LEGAL_ENTITY_ACTOR", "mustermann holding gmbh", " Mustermann Holding GmbH "], ) def test_switch_account_finds_the_account_by_id_type_or_name(account): - tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr = _api([BOTH, {}]) assert tr.switch_account(account) == COMPANY +@pytest.mark.parametrize("account", ["Erika", "erika mustermann", "ADULT", "self"]) +def test_switch_account_finds_a_person_by_first_or_full_name(account): + tr = _api([BOTH, {}]) + + assert tr.switch_account(account) == SELF + + +@pytest.mark.parametrize("account", ["Mustermann", "Holding", "LEGAL", "id-", ""]) +def test_switch_account_does_not_match_on_parts_of_a_name(account): + tr = _api([BOTH]) + + with pytest.raises(ValueError, match="was not found"): + tr.switch_account(account) + + def test_switch_account_without_a_name_returns_to_the_own_account(): - tr = _api([{"relationships": [SELF, COMPANY]}, {}]) - tr._subject_id = "id-company" + tr = _switched([BOTH, {}]) assert tr.switch_account() == SELF assert tr._websession.calls[1]["json"] == {"subjectId": "id-self"} assert tr._subject_id is None +def test_switch_account_reports_a_login_without_an_own_account(): + tr = _api([{"relationships": [COMPANY]}]) + + with pytest.raises(ValueError, match="Your own account was not found"): + tr.switch_account() + + def test_switch_account_rejects_an_unknown_account_and_names_the_known_ones(): - tr = _api([{"relationships": [SELF, COMPANY]}]) + tr = _api([BOTH]) - with pytest.raises(ValueError, match="does not match any account.*LEGAL_ENTITY \\(Mustermann Holding GmbH\\)"): + with pytest.raises(ValueError, match="'nope' was not found.*LEGAL_ENTITY \\(Mustermann Holding GmbH\\)"): tr.switch_account("nope") assert _calls(tr) == [("GET", RELATIONSHIPS)] -def test_switch_account_rejects_an_ambiguous_account(): - """Two children share their type, so the type alone does not say which one is meant.""" - tr = _api([{"relationships": [SELF, CHILD_A, CHILD_B]}, {"relationships": [SELF, CHILD_A, CHILD_B]}, {}]) +def test_switch_account_rejects_an_ambiguous_account_until_given_its_id(): + """Two companies share their type, so the type alone does not say which one is meant.""" + accounts = {"relationships": [SELF, COMPANY, SECOND_COMPANY]} + tr = _api([accounts, accounts, {}]) + + with pytest.raises(ValueError, match="more than one account, use its customer id"): + tr.switch_account("LEGAL_ENTITY") + + assert tr.switch_account("id-second") == SECOND_COMPANY + + +def test_switch_account_rejects_an_account_without_a_customer_id(): + tr = _api([{"relationships": [SELF, {"accountType": "LEGAL_ENTITY", "accountName": "Nameless"}]}]) + + with pytest.raises(ValueError, match="was not found"): + tr.switch_account("LEGAL_ENTITY") + - with pytest.raises(ValueError, match="more than one account"): - tr.switch_account("CHILD") +def test_switch_account_copes_with_fields_that_are_not_text(): + odd = {"customerId": 42, "firstName": 1, "lastName": None, "relationshipType": "LEGAL_ENTITY_ACTOR"} + tr = _api([{"relationships": [SELF, odd]}, {}]) - assert tr.switch_account("Mia") == CHILD_B + assert tr.switch_account("42") == odd -def test_switch_account_reports_a_refused_switch(): - tr = _api([{"relationships": [SELF, COMPANY]}, (403, {})]) +def test_a_refused_switch_leaves_the_session_as_it_was(): + tr = _api([BOTH, (403, {})]) + tr._sec_acc_no = "own-depot" + tr._session_expires_at = expires_at = 10**10 + tr.save_websession = lambda: pytest.fail("must not save the cookies of a refused switch") with pytest.raises(requests.HTTPError): tr.switch_account("LEGAL_ENTITY") - assert tr._subject_id is None + assert (tr._subject_id, tr._sec_acc_no, tr._session_expires_at) == (None, "own-depot", expires_at) def test_switch_account_refuses_while_the_websocket_is_connected(): """An open websocket keeps answering for the account it was authenticated with.""" - tr = _api([{"relationships": [SELF, COMPANY]}, {}]) + tr = _api([]) tr._ws = _Websocket({}) with pytest.raises(ValueError, match="websocket"): @@ -216,48 +291,35 @@ def test_switch_account_refuses_while_the_websocket_is_connected(): # --- keeping the session on the chosen account --------------------------------------- -def test_session_refresh_stays_on_the_chosen_account(): - tr = _api([{"relationships": [SELF, COMPANY]}, {}, {}, {}]) - tr.switch_account("LEGAL_ENTITY") - tr._session_expires_at = 0 +def test_the_switch_counts_as_a_session_refresh(): + tr = _api([BOTH, {}, {}]) + tr._session_expires_at = 10**10 # far enough ahead that only the switch can have moved it + tr.switch_account("LEGAL_ENTITY") tr._web_request("/api/v1/whatever") - assert _calls(tr)[2] == ("POST", SESSION_V2) - assert tr._websession.calls[2]["json"] == {"subjectId": "id-company"} + assert tr._session_expires_at < 10**10 + assert _calls(tr) == [("GET", RELATIONSHIPS), ("POST", SESSION_V2), ("GET", f"{HOST}/api/v1/whatever")] -def test_session_refresh_is_unchanged_for_the_own_account(): - tr = _api([{}, {}], session_is_fresh=False) +def test_session_refresh_stays_on_the_chosen_account(): + """A refresh without the subject would silently return the session to the own account.""" + tr = _switched([{}, {}]) + tr._session_expires_at = 0 tr._web_request("/api/v1/whatever") - assert _calls(tr)[0] == ("GET", SESSION_V1) + assert _calls(tr) == [("POST", SESSION_V2), ("GET", f"{HOST}/api/v1/whatever")] + assert tr._websession.calls[0]["json"] == {"subjectId": "id-company"} -def _resumable(tmp_path, replies): - cookies = tmp_path / "cookies.txt" - cookies.write_text("# Netscape HTTP Cookie File\n") - tr = TradeRepublicApi( - phone_no="+490000000000", pin="0000", waf_token=None, save_cookies=True, cookies_file=str(cookies) - ) - jar = tr._websession.cookies - tr._websession = _Session(replies) - tr._websession.cookies = jar # type: ignore[attr-defined] - return tr - - -def test_resume_accepts_cookies_saved_while_acting_for_another_account(tmp_path): - """Those cookies are valid; only the account settings are not served for them.""" - tr = _resumable(tmp_path, [{}, OTHER_ACCOUNT]) - - assert tr.resume_websession() is True - +def test_a_new_process_starts_on_the_own_account(): + """Its first request refreshes without a subject, whatever account the saved cookies act for.""" + tr = _api([{}, {}], session_is_fresh=False) -def test_resume_still_rejects_an_expired_session(tmp_path): - tr = _resumable(tmp_path, [(401, {})]) + tr.settings() - assert tr.resume_websession() is False + assert _calls(tr) == [("GET", SESSION_V1), ("GET", ACCOUNT)] # --- portfolio ----------------------------------------------------------------------- @@ -272,9 +334,8 @@ def test_resume_still_rejects_an_expired_session(tmp_path): def test_compact_portfolio_reads_the_securities_account_from_account_pairs_after_a_switch(): - tr = _api([{"relationships": [SELF, COMPANY]}, {}]) - tr.switch_account("LEGAL_ENTITY") - ws = _with_websocket(tr, {"accountPairs": PAIRS}) + tr = _switched() + ws = _with_websocket(tr, {"accountPairs": [("A", "accountPairs", PAIRS)]}) asyncio.run(tr.compact_portfolio()) @@ -282,54 +343,96 @@ def test_compact_portfolio_reads_the_securities_account_from_account_pairs_after {"type": "accountPairs"}, {"type": "compactPortfolioByType", "secAccNo": "depot-company"}, ] - # The account settings are not served for this account, so they are not asked for. - assert ("GET", ACCOUNT) not in _calls(tr) + # The account settings are refused for this account, so they are not asked for. + assert _calls(tr) == [] + assert "unsub 1" in ws.sent -def test_compact_portfolio_falls_back_to_account_pairs_when_the_settings_are_refused(): - tr = _api([OTHER_ACCOUNT]) - ws = _with_websocket(tr, {"accountPairs": PAIRS}) +def test_compact_portfolio_takes_the_first_pair_when_none_is_the_default_product(): + tr = _switched() + pairs = {"accounts": [{"cashAccountNumber": "c"}, {"securitiesAccountNumber": "depot-a", "productType": "X"}]} + ws = _with_websocket(tr, {"accountPairs": [("A", "accountPairs", pairs)]}) asyncio.run(tr.compact_portfolio()) - assert ws.subscriptions()[-1] == {"type": "compactPortfolioByType", "secAccNo": "depot-company"} + assert ws.subscriptions()[-1]["secAccNo"] == "depot-a" -def test_compact_portfolio_still_uses_the_account_settings_for_the_own_account(): - tr = _api([{"securitiesAccountNumber": "own-depot"}]) - ws = _with_websocket(tr, {}) +def test_compact_portfolio_keeps_the_answers_to_other_subscriptions(): + """Looking up the securities account must not swallow what the caller subscribed to before.""" + tr = _switched() + _with_websocket( + tr, + { + "cash": [], + "accountPairs": [("A", "cash", [{"amount": 1}]), ("A", "accountPairs", PAIRS)], + "compactPortfolioByType": [("A", "compactPortfolioByType", {"categories": []})], + }, + ) - asyncio.run(tr.compact_portfolio()) + async def cash_then_portfolio(): + await tr.cash() + await tr.compact_portfolio() + return [await tr.recv(), await tr.recv()] - assert ws.subscriptions() == [{"type": "compactPortfolioByType", "secAccNo": "own-depot"}] + received = asyncio.run(asyncio.wait_for(cash_then_portfolio(), 5)) + assert [(subscription["type"], payload) for _, subscription, payload in received] == [ + ("cash", [{"amount": 1}]), + ("compactPortfolioByType", {"categories": []}), + ] -def test_compact_portfolio_does_not_hide_other_errors_of_the_settings_call(): - tr = _api([(500, {})]) - _with_websocket(tr, {"accountPairs": PAIRS}) - with pytest.raises(requests.HTTPError): +@pytest.mark.parametrize( + "answers", + [ + [], # no answer at all + [("A", "accountPairs", {"authAccountId": "id-company", "accounts": []})], + [("A", "accountPairs", {"authAccountId": "id-company", "accounts": None})], + ], +) +def test_compact_portfolio_reports_a_missing_securities_account_after_a_switch(answers, monkeypatch): + tr = _switched() + ws = _with_websocket(tr, {"accountPairs": answers}) + lookup = tr._sec_acc_no_from_account_pairs + monkeypatch.setattr(tr, "_sec_acc_no_from_account_pairs", lambda: lookup(timeout=0.05)) + + with pytest.raises(ValueError, match="securities account number from account pairs"): asyncio.run(tr.compact_portfolio()) + assert "unsub 1" in ws.sent + -def test_compact_portfolio_reports_a_missing_securities_account(): +def test_compact_portfolio_is_unchanged_for_the_own_account(): + tr = _api([{"securitiesAccountNumber": "own-depot"}]) + ws = _with_websocket(tr, {}) + + asyncio.run(tr.compact_portfolio()) + + assert _calls(tr) == [("GET", ACCOUNT)] + assert ws.subscriptions() == [{"type": "compactPortfolioByType", "secAccNo": "own-depot"}] + + +def test_compact_portfolio_still_reports_account_settings_without_a_securities_account(): tr = _api([{}]) - _with_websocket(tr, {"accountPairs": {"authAccountId": "id-self", "accounts": []}}) + ws = _with_websocket(tr, {}) - with pytest.raises(ValueError, match="securities account number"): + with pytest.raises(ValueError, match="securities account number from account settings"): asyncio.run(tr.compact_portfolio()) + assert ws.sent == [] + # --- pytr accounts ------------------------------------------------------------------- def test_get_accounts_prints_what_account_accepts(): - tr = _api([{"relationships": [SELF, {**COMPANY, "accountState": "ACTIVE"}, CHILD_B]}]) + tr = _api([{"relationships": [SELF, COMPANY, {"customerId": "id-bare"}]}]) assert account_module.get_accounts(tr).splitlines() == [ - "ADULT Erika Mustermann your own account", - "LEGAL_ENTITY Mustermann Holding GmbH ACTIVE", - "CHILD Mia", + "ADULT Erika Mustermann id-self your own account", + "LEGAL_ENTITY Mustermann Holding GmbH id-company ACTIVE", + "UNKNOWN id-bare", ] @@ -338,57 +441,70 @@ def test_get_accounts_of_a_login_without_relationships_is_empty(): def test_the_accounts_command_exists_and_takes_the_login_arguments(): - from pytr.main import get_main_parser - args = get_main_parser().parse_args(["accounts", "--v2"]) assert (args.command, args.v2, args.account) == ("accounts", True, None) +def test_every_command_that_logs_in_takes_an_account(): + parser = get_main_parser() + + for command in (["login"], ["portfolio"], ["details", "DE0000000000"], ["dl_docs", "out"], ["get_price_alarms"]): + assert parser.parse_args([*command, "--account", "LEGAL_ENTITY"]).account == "LEGAL_ENTITY" + + # --- login() ------------------------------------------------------------------------- class _LoggedIn: """Stands in for a TradeRepublicApi whose saved session could be resumed.""" - def __init__(self, settings_reply): - self._settings_reply = settings_reply - self.switched_to: list[Any] = [] + def __init__(self, switch_error=None): + self._switch_error = switch_error + self.events: list[Any] = [] def resume_websession(self): return True def settings(self): - status, payload = self._settings_reply - response = _Response(payload, status) - response.raise_for_status() - return payload - - _acts_for_other_account = staticmethod(TradeRepublicApi._acts_for_other_account) + self.events.append("settings") + return {} def switch_account(self, account=None): - self.switched_to.append(account) + self.events.append(("switch", account)) + if self._switch_error: + raise self._switch_error -def _login(monkeypatch, settings_reply, **kwargs): - tr = _LoggedIn(settings_reply) +def _login(monkeypatch, tmp_path, tr, **kwargs): + # login() creates the directory it stores credentials in; keep it off the real one. + monkeypatch.setattr(account_module, "BASE_DIR", tmp_path / "pytr") + monkeypatch.setattr(account_module, "CREDENTIALS_FILE", tmp_path / "pytr" / "credentials") monkeypatch.setattr(account_module, "TradeRepublicApi", lambda **_: tr) - assert account_module.login(phone_no="+490000000000", pin="0000", **kwargs) is tr - return tr + return account_module.login(phone_no="+490000000000", pin="0000", **kwargs) + +def test_login_does_not_switch_by_default(monkeypatch, tmp_path): + tr = _LoggedIn() -def test_login_does_not_switch_by_default(monkeypatch): - assert _login(monkeypatch, (200, {})).switched_to == [] + assert _login(monkeypatch, tmp_path, tr) is tr + assert tr.events == ["settings"] -def test_login_switches_to_the_requested_account(monkeypatch): - assert _login(monkeypatch, (200, {}), account="LEGAL_ENTITY").switched_to == ["LEGAL_ENTITY"] +def test_login_switches_to_the_requested_account(monkeypatch, tmp_path): + tr = _LoggedIn() + assert _login(monkeypatch, tmp_path, tr, account="LEGAL_ENTITY") is tr + # The settings are read first: they are refused once the session acts for the other account. + assert tr.events == ["settings", ("switch", "LEGAL_ENTITY")] -def test_login_stays_on_the_requested_account_when_the_cookies_already_act_for_it(monkeypatch): - assert _login(monkeypatch, OTHER_ACCOUNT, account="LEGAL_ENTITY").switched_to == ["LEGAL_ENTITY"] +@pytest.mark.parametrize( + "error", + [ValueError("'nope' was not found"), requests.HTTPError("403", response=_Response({}, 403))], # type: ignore[arg-type] +) +def test_login_exits_when_the_account_cannot_be_used(monkeypatch, tmp_path, error): + with pytest.raises(SystemExit) as excinfo: + _login(monkeypatch, tmp_path, _LoggedIn(switch_error=error), account="nope") -def test_login_returns_to_the_own_account_when_none_is_requested(monkeypatch): - """Cookies saved by a run with --account must not make the next plain run use that account.""" - assert _login(monkeypatch, OTHER_ACCOUNT).switched_to == [None] + assert excinfo.value.code == 1 From 0989609b47f6db7696912ae0c72b183e709193bf Mon Sep 17 00:00:00 2001 From: Christoph Langer Date: Sat, 3 Oct 2026 14:43:26 +0200 Subject: [PATCH 4/5] Rename get_accounts -> print_accounts add heading line to print_accounts minor fix for using v2 argument in main.py --- pytr/account.py | 10 +++++++--- pytr/main.py | 6 ++++-- tests/test_accounts.py | 13 +++++++------ 3 files changed, 18 insertions(+), 11 deletions(-) diff --git a/pytr/account.py b/pytr/account.py index 1c322695..b5203fe4 100644 --- a/pytr/account.py +++ b/pytr/account.py @@ -20,7 +20,7 @@ def get_settings(tr): return formatted_json -def get_accounts(tr): +def print_accounts(tr): """List the accounts of this login, one per line: type, name, customer id and state. `--account` takes the type, the name or, where those are not unique, the customer id. @@ -34,8 +34,12 @@ def get_accounts(tr): ) for r in tr.relationships() ] - widths = [max((len(row[i]) for row in rows), default=0) for i in range(3)] - return "\n".join(" ".join(f"{cell:<{width}}" for cell, width in zip(row, widths + [0])).rstrip() for row in rows) + if not rows: + return "" + heading = ("TYPE", "NAME", "CUSTOMER ID", "STATE") + widths = [max(len(heading[i]), max(len(row[i]) for row in rows)) for i in range(4)] + all_rows = [heading] + rows + return "\n".join(" ".join(f"{cell:<{width}}" for cell, width in zip(row, widths)).rstrip() for row in all_rows) def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False, account=None): diff --git a/pytr/main.py b/pytr/main.py index d13a5f7b..41413dac 100644 --- a/pytr/main.py +++ b/pytr/main.py @@ -12,7 +12,7 @@ import shtab -from pytr.account import get_accounts, login +from pytr.account import login, print_accounts from pytr.alarms import Alarms from pytr.details import Details from pytr.dl import DL @@ -583,7 +583,7 @@ def main(): ) elif args.command == "accounts": print( - get_accounts( + print_accounts( login( phone_no=args.phone_no, pin=args.pin, @@ -726,6 +726,7 @@ def main(): pin=args.pin, store_credentials=args.store_credentials, waf_token=args.waf_token, + v2=args.v2, account=args.account, ), args.input, @@ -742,6 +743,7 @@ def main(): pin=args.pin, store_credentials=args.store_credentials, waf_token=args.waf_token, + v2=args.v2, account=args.account, ), args.input, diff --git a/tests/test_accounts.py b/tests/test_accounts.py index 2ca791ba..5c843f4b 100644 --- a/tests/test_accounts.py +++ b/tests/test_accounts.py @@ -426,18 +426,19 @@ def test_compact_portfolio_still_reports_account_settings_without_a_securities_a # --- pytr accounts ------------------------------------------------------------------- -def test_get_accounts_prints_what_account_accepts(): +def test_print_accounts_prints_what_account_accepts(): tr = _api([{"relationships": [SELF, COMPANY, {"customerId": "id-bare"}]}]) - assert account_module.get_accounts(tr).splitlines() == [ - "ADULT Erika Mustermann id-self your own account", - "LEGAL_ENTITY Mustermann Holding GmbH id-company ACTIVE", + assert account_module.print_accounts(tr).splitlines() == [ + "TYPE NAME CUSTOMER ID STATE", + "ADULT Erika Mustermann id-self your own account", + "LEGAL_ENTITY Mustermann Holding GmbH id-company ACTIVE", "UNKNOWN id-bare", ] -def test_get_accounts_of_a_login_without_relationships_is_empty(): - assert account_module.get_accounts(_api([{}])) == "" +def test_print_accounts_of_a_login_without_relationships_is_empty(): + assert account_module.print_accounts(_api([{}])) == "" def test_the_accounts_command_exists_and_takes_the_login_arguments(): From 804d667326c39f56abd121148cb2a0553ce543ea Mon Sep 17 00:00:00 2001 From: Christoph Langer Date: Wed, 7 Oct 2026 22:48:15 +0200 Subject: [PATCH 5/5] Improve documentation --- README.md | 15 +++++++-------- pytr/main.py | 2 +- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 448150cc..b2d711a1 100644 --- a/README.md +++ b/README.md @@ -72,8 +72,8 @@ Commands: help Print this help message login Check if credentials file exists. If not create it and ask for input. Try to login. Ask for device reset if needed - accounts List the accounts of this login, e.g. a company account. Use one of them with - --account + accounts List accounts of this login, e.g. a company accounts or child accounts. Use them + with --account portfolio Show current portfolio rates Fetch current prices for a list of ISINs given as direct list or CSV input details Get details for an ISIN @@ -121,9 +121,9 @@ running `pytr`. ### Other accounts of your login -A Trade Republic login can hold more than its own account, e.g. a company account. Trade Republic treats each of -them as a customer of its own, and the web app switches between them after the login. Pass `--account` to do the -same, on any subcommand that performs a login: +A Trade Republic login can hold more than its own account, e.g. company accounts or child accounts. Trade Republic +treats each of them as a customer of its own, and the web app switches between them after the login. Pass `--account` +to do the same, on any subcommand that performs a login: ```sh pytr portfolio --account LEGAL_ENTITY @@ -131,10 +131,9 @@ pytr dl_docs --account "Mustermann Holding GmbH" ./docs ``` The value is the account's type, its name or its customer id, as listed by `pytr accounts`. Without `--account`, -pytr uses your own account, as before. The choice is not remembered between runs. +pytr uses your own account. The choice is not remembered between runs. -This has been tested with a company account. Other accounts that the web app lets you switch to, e.g. a child's -account, are expected to work the same way but have not been tested. +Unfortunately, it seems that this only works with company accounts, not with child accounts at the moment. ## Development diff --git a/pytr/main.py b/pytr/main.py index 41413dac..d994f567 100644 --- a/pytr/main.py +++ b/pytr/main.py @@ -176,7 +176,7 @@ def formatter(prog): ) # accounts - info = "List the accounts of this login, e.g. a company account. Use one of them with --account" + info = "List accounts of this login, e.g. a company accounts or child accounts. Use them with --account" parser_cmd.add_parser( "accounts", formatter_class=formatter,