From 0820c11919908e914693cff5e6507ed74c9b6745 Mon Sep 17 00:00:00 2001 From: Ludy Date: Tue, 9 Jun 2026 11:00:32 +0200 Subject: [PATCH] Update files Signed-off-by: GitHub Action --- .github/workflows/dependency-review.yml | 6 +- .github/workflows/labeler.yml | 4 +- .github/workflows/scorecards.yml | 2 +- src/pyxplora_api/const.py | 4 +- src/pyxplora_api/exception_classes.py | 18 +-- src/pyxplora_api/gql_handler.py | 162 +++++-------------- src/pyxplora_api/gql_handler_async.py | 206 ++++++------------------ src/pyxplora_api/gql_queries.py | 4 +- src/pyxplora_api/graphql_client.py | 24 +-- src/pyxplora_api/handler_gql.py | 8 +- src/pyxplora_api/model.py | 16 +- src/pyxplora_api/pyxplora.py | 40 ++--- src/pyxplora_api/pyxplora_api.py | 109 +++---------- src/pyxplora_api/pyxplora_api_async.py | 137 ++++------------ tests/test_exception_classes.py | 13 +- tests/test_gql_handler_sync.py | 30 +--- tests/test_graphql_client.py | 20 +-- tests/test_handler_gql.py | 6 +- tests/test_models_and_status.py | 4 +- tests/test_pyxplora_core.py | 8 +- 20 files changed, 207 insertions(+), 614 deletions(-) diff --git a/.github/workflows/dependency-review.yml b/.github/workflows/dependency-review.yml index 23a9a8f..9c76845 100644 --- a/.github/workflows/dependency-review.yml +++ b/.github/workflows/dependency-review.yml @@ -6,7 +6,7 @@ # PRs introducing known-vulnerable packages will be blocked from merging. # # Source repository: https://github.com/actions/dependency-review-action -name: 'Dependency Review' +name: "Dependency Review" on: [pull_request] permissions: @@ -21,7 +21,7 @@ jobs: with: egress-policy: audit - - name: 'Checkout Repository' + - name: "Checkout Repository" uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 - - name: 'Dependency Review' + - name: "Dependency Review" uses: actions/dependency-review-action@a1d282b36b6f3519aa1f3fc636f609c47dddb294 # v5.0.0 diff --git a/.github/workflows/labeler.yml b/.github/workflows/labeler.yml index cfdd383..6af213f 100644 --- a/.github/workflows/labeler.yml +++ b/.github/workflows/labeler.yml @@ -14,8 +14,8 @@ permissions: jobs: labeler: permissions: - contents: read # for actions/checkout to fetch code - issues: write # for crazy-max/ghaction-github-labeler to create, rename, update and delete label + contents: read # for actions/checkout to fetch code + issues: write # for crazy-max/ghaction-github-labeler to create, rename, update and delete label name: Labeler runs-on: ubuntu-latest steps: diff --git a/.github/workflows/scorecards.yml b/.github/workflows/scorecards.yml index 7090c5a..05228a8 100644 --- a/.github/workflows/scorecards.yml +++ b/.github/workflows/scorecards.yml @@ -10,7 +10,7 @@ on: # To guarantee Maintained check is occasionally updated. See # https://github.com/ossf/scorecard/blob/main/docs/checks.md#maintained schedule: - - cron: '20 7 * * 2' + - cron: "20 7 * * 2" push: branches: ["main"] diff --git a/src/pyxplora_api/const.py b/src/pyxplora_api/const.py index cde5f4d..5c2a385 100644 --- a/src/pyxplora_api/const.py +++ b/src/pyxplora_api/const.py @@ -2,4 +2,6 @@ API_SECRET = "1e9b6fe0327711ed959359c157878dcb" ENDPOINT = "https://api.myxplora.com/api" DEFAULT_TIMEOUT = 60 -DEFAULT_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.3" +DEFAULT_USER_AGENT = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.3" +) diff --git a/src/pyxplora_api/exception_classes.py b/src/pyxplora_api/exception_classes.py index a9c3299..cd62aa2 100644 --- a/src/pyxplora_api/exception_classes.py +++ b/src/pyxplora_api/exception_classes.py @@ -41,11 +41,7 @@ class ChildNoError(Error): """Exception raised when a child's phone number or watch ID is not found.""" def __init__(self, error_message=None) -> None: - error_message = ( - ["Child phonenumber", "Watch ID"] - if error_message is None - else error_message - ) + error_message = ["Child phonenumber", "Watch ID"] if error_message is None else error_message self.error_message = error_message super().__init__() @@ -63,7 +59,9 @@ def __init__(self, allow, deny) -> None: super().__init__() def __str__(self) -> str: - return f"Transfer value has the wrong type! The following are permitted: {self.allow}. The specified type is: {self.deny}" + return ( + f"Transfer value has the wrong type! The following are permitted: {self.allow}. The specified type is: {self.deny}" + ) class FunctionError(Error): @@ -82,9 +80,7 @@ class LoginError(Error): """Exception raised when login to the Xplora API fails.""" def __init__(self, error_message: str | ErrorMSG = "") -> None: - self.error_message = ( - error_message if isinstance(error_message, str) else error_message.value - ) + self.error_message = error_message if isinstance(error_message, str) else error_message.value super().__init__() def __str__(self) -> str: @@ -95,9 +91,7 @@ class PhoneOrEmailFail(Error): """Exception raised when phone number or email address is not found.""" def __init__(self, error_message: str | ErrorMSG = ErrorMSG.PHONE_MAIL_ERR) -> None: - self.error_message = ( - error_message if isinstance(error_message, str) else error_message.value - ) + self.error_message = error_message if isinstance(error_message, str) else error_message.value super().__init__() def __str__(self) -> str: diff --git a/src/pyxplora_api/gql_handler.py b/src/pyxplora_api/gql_handler.py index 9a59f88..125b7fa 100644 --- a/src/pyxplora_api/gql_handler.py +++ b/src/pyxplora_api/gql_handler.py @@ -39,9 +39,7 @@ def __init__( email: str | None = None, signup: bool = True, ) -> None: - super().__init__( - countryPhoneNumber, phoneNumber, password, userLang, timeZone, email, signup - ) + super().__init__(countryPhoneNumber, phoneNumber, password, userLang, timeZone, email, signup) def runGqlQuery( self, @@ -69,9 +67,7 @@ def runGqlQuery( # create GQLClient gqlClient = GraphqlClient(endpoint=ENDPOINT, headers=requestHeaders) # execute QUERY|MUTATION - data: dict[str, Any] = gqlClient.execute( - query=query, variables=variables, operation_name=operation_name - ) + data: dict[str, Any] = gqlClient.execute(query=query, variables=variables, operation_name=operation_name) return data def runAuthorizedGqlQuery( @@ -122,9 +118,7 @@ def login(self) -> dict[str, Any]: data = dataAll.get("data", {}) signIn = data.get("signInWithEmailOrPhone", None) if signIn is None: - error_message = dataAll.get("errors", [{"message": ""}])[0].get( - "message", "" - ) + error_message = dataAll.get("errors", [{"message": ""}])[0].get("message", "") raise LoginError(f"Login error: {error_message}") self.issueToken = signIn @@ -134,9 +128,7 @@ def login(self) -> dict[str, Any]: return self.issueToken - def isAdmin( - self, wuid: str, query: str, variables: dict[str, Any], key: str - ) -> bool: + def isAdmin(self, wuid: str, query: str, variables: dict[str, Any], key: str) -> bool: """This method determines whether the currently logged-in user is an admin for the specified watch user. Args: @@ -160,9 +152,7 @@ def isAdmin( id = None if self.userId == id: if contact["guardianType"] == "FIRST": - data: dict[str, Any] = self.runAuthorizedGqlQuery( - query, variables, key - ).get("data", {}) + data: dict[str, Any] = self.runAuthorizedGqlQuery(query, variables, key).get("data", {}) for k in data: if k.upper() == key.upper(): return data.get(k, False) @@ -179,9 +169,7 @@ def askWatchLocate(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: A dictionary containing the response of the query, with a key "askWatchLocate". """ - data: dict[str, Any] = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("askLocateQ", ""), {"uid": wuid}, "AskWatchLocate" - ) + data: dict[str, Any] = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("askLocateQ", ""), {"uid": wuid}, "AskWatchLocate") errors = data.get("errors", []) if errors: self.errors.append({"function": "askWatchLocate", "errors": errors}) @@ -199,9 +187,7 @@ def getWatchUserContacts(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: A dictionary containing the response of the query. """ - data: dict[str, Any] = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("contactsQ", ""), {"uid": wuid}, "Contacts" - ) + data: dict[str, Any] = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("contactsQ", ""), {"uid": wuid}, "Contacts") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatchUserContacts", "errors": errors}) @@ -216,9 +202,7 @@ def getWatches(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: A dictionary containing the response of the query. """ - data: dict[str, Any] = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("watchesQ", ""), {"uid": wuid}, "Watches" - ) + data: dict[str, Any] = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("watchesQ", ""), {"uid": wuid}, "Watches") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatches", "errors": errors}) @@ -243,9 +227,7 @@ def getSWInfo(self, qrCode: str) -> dict[str, Any]: self.errors.append({"function": "getSWInfo", "errors": errors}) return data.get("data", {}) - def getWatchState( - self, qrCode: str, qrt: str = "", qrc: str = "" - ) -> dict[str, Any]: + def getWatchState(self, qrCode: str, qrt: str = "", qrc: str = "") -> dict[str, Any]: """Get the state of a watch using its QR code. Args: @@ -263,9 +245,7 @@ def getWatchState( variables["qrt"] = qrt if qrc: variables["qrc"] = qrc - data: dict[str, Any] = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("stateQ", ""), variables, "WatchState" - ) + data: dict[str, Any] = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("stateQ", ""), variables, "WatchState") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatchState", "errors": errors}) @@ -280,9 +260,7 @@ def getWatchLastLocation(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: A dictionary containing the response of the query. """ - data: dict[str, Any] = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("locateQ", ""), {"uid": wuid}, "WatchLastLocate" - ) + data: dict[str, Any] = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("locateQ", ""), {"uid": wuid}, "WatchLastLocate") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatchLastLocation", "errors": errors}) @@ -297,9 +275,7 @@ def trackWatch(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: A dictionary containing the response of the query. """ - data: dict[str, Any] = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("trackQ", ""), {"uid": wuid}, "TrackWatch" - ) + data: dict[str, Any] = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("trackQ", ""), {"uid": wuid}, "TrackWatch") errors = data.get("errors", []) if errors: self.errors.append({"function": "trackWatch", "errors": errors}) @@ -317,9 +293,7 @@ def getAlarmTime(self, wuid: str) -> dict[str, Any]: Returns: dict: A dictionary containing the response of the query. """ - return self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("alarmsQ", ""), {"uid": wuid}, "Alarms" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCH_Q.get("alarmsQ", ""), {"uid": wuid}, "Alarms").get("data", {}) def getWifi(self, wuid: str) -> dict[str, Any]: """Get the Wi-Fi information of a watch. @@ -330,9 +304,7 @@ def getWifi(self, wuid: str) -> dict[str, Any]: Returns: dict: A dictionary containing the response of the query. """ - return self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("getWifisQ", ""), {"uid": wuid}, "GetWifis" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCH_Q.get("getWifisQ", ""), {"uid": wuid}, "GetWifis").get("data", {}) def unReadChatMsgCount(self, wuid: str) -> dict[str, Any]: """Get the count of unread chat messages for a watch. @@ -358,9 +330,7 @@ def safeZones(self, wuid: str) -> dict[str, Any]: Returns: dict: A dictionary containing the response of the query. """ - return self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("safeZonesQ", ""), {"uid": wuid}, "SafeZones" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCH_Q.get("safeZonesQ", ""), {"uid": wuid}, "SafeZones").get("data", {}) def safeZoneGroups(self) -> dict[str, Any]: """Get the safe zone groups. @@ -368,9 +338,7 @@ def safeZoneGroups(self) -> dict[str, Any]: Returns: dict: A dictionary containing the response of the query. """ - return self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("safeZoneGroupsQ", ""), {}, "SafeZoneGroups" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCH_Q.get("safeZoneGroupsQ", ""), {}, "SafeZoneGroups").get("data", {}) def silentTimes(self, wuid: str) -> dict[str, Any]: """Get the silent times for a watch. @@ -381,9 +349,7 @@ def silentTimes(self, wuid: str) -> dict[str, Any]: Returns: dict: A dictionary containing the response of the query. """ - return self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("silentTimesQ", ""), {"uid": wuid}, "SlientTimes" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCH_Q.get("silentTimesQ", ""), {"uid": wuid}, "SlientTimes").get("data", {}) def chats( self, @@ -516,9 +482,7 @@ def watchImei(self, imei: str, qrCode: str, deviceKey: str) -> dict[str, Any]: "WatchImei", ).get("data", {}) - def getWatchLocHistory( - self, wuid: str, date: int, tz: str, limit: int - ) -> dict[str, Any]: + def getWatchLocHistory(self, wuid: str, date: int, tz: str, limit: int) -> dict[str, Any]: """Retrieve the location history for a watch with a given WUID. Args: @@ -542,13 +506,9 @@ def watchesDynamic(self) -> dict[str, Any]: Returns: dict[str, Any]: Dynamic data for all watches. """ - return self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("watchesDynamicQ", ""), {}, "WatchesDynamic" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCH_Q.get("watchesDynamicQ", ""), {}, "WatchesDynamic").get("data", {}) - def coinHistory( - self, wuid: str, start: int, end: int, type: str, offset: int, limit: int - ) -> dict[str, Any]: + def coinHistory(self, wuid: str, start: int, end: int, type: str, offset: int, limit: int) -> dict[str, Any]: """Retrieve coin history for a watch with a given WUID. Args: @@ -584,9 +544,7 @@ def reminders(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: The reminder data for the given user. """ - return self.runAuthorizedGqlQuery( - gq.XMOVE_Q.get("remindersQ", ""), {"uid": wuid}, "Reminders" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.XMOVE_Q.get("remindersQ", ""), {"uid": wuid}, "Reminders").get("data", {}) def groups(self, isCampaign: bool) -> dict[str, Any]: """Get card group data. @@ -597,9 +555,9 @@ def groups(self, isCampaign: bool) -> dict[str, Any]: Returns: dict[str, Any]: The card group data. """ - return self.runAuthorizedGqlQuery( - gq.CARD_Q.get("groupsQ", ""), {"isCampaign": isCampaign}, "CardGroups" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.CARD_Q.get("groupsQ", ""), {"isCampaign": isCampaign}, "CardGroups").get( + "data", {} + ) def dynamic(self) -> dict[str, Any]: """Get dynamic card data. @@ -607,9 +565,7 @@ def dynamic(self) -> dict[str, Any]: Returns: dict[str, Any]: The dynamic card data. """ - return self.runAuthorizedGqlQuery( - gq.CARD_Q.get("dynamicQ", ""), {}, "DynamicCards" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.CARD_Q.get("dynamicQ", ""), {}, "DynamicCards").get("data", {}) def staticCard(self) -> dict[str, Any]: """Get dynamic card data. @@ -617,9 +573,7 @@ def staticCard(self) -> dict[str, Any]: Returns: dict[str, Any]: The dynamic card data. """ - return self.runAuthorizedGqlQuery( - gq.CARD_Q.get("staticQ", ""), {}, "StaticCard" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.CARD_Q.get("staticQ", ""), {}, "StaticCard").get("data", {}) def familyInfo(self, wuid: str, watchId: str, tz: str, date: int) -> dict[str, Any]: """Get family information for a given user. @@ -725,9 +679,7 @@ def getMyInfo(self) -> dict[str, Any]: dict: Information retrieved for the logged in user, including the data field. """ # Profil from login Account - return self.runAuthorizedGqlQuery( - gq.MYINFO_Q.get("readQ", ""), {}, "ReadMyInfo" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.MYINFO_Q.get("readQ", ""), {}, "ReadMyInfo").get("data", {}) def readCampaignProfile(self, wuid: str) -> dict[str, Any]: """Retrieve campaign profile for the given user. @@ -752,9 +704,7 @@ def getReviewStatus(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: The review status data in dictionary format. """ - return self.runAuthorizedGqlQuery( - gq.REVIEW_Q.get("getStatusQ", ""), {"uid": wuid}, "GetReviewStatus" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.REVIEW_Q.get("getStatusQ", ""), {"uid": wuid}, "GetReviewStatus").get("data", {}) def getWatchUserSteps(self, wuid: str, tz: str, date: int) -> dict[str, Any]: """Get the step count data for a given user id. @@ -783,9 +733,7 @@ def countries(self) -> dict[str, Any]: Returns: dict[str, Any]: The list of countries in dictionary format. """ - return self.runAuthorizedGqlQuery( - gq.UTILS_Q.get("countriesQ", ""), {}, "Countries" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.UTILS_Q.get("countriesQ", ""), {}, "Countries").get("data", {}) def avatars(self, id: str) -> dict[str, Any]: """Get the avatar data for a given id. @@ -796,9 +744,7 @@ def avatars(self, id: str) -> dict[str, Any]: Returns: dict[str, Any]: The avatar data in dictionary format. """ - return self.runAuthorizedGqlQuery( - gq.CAMPAIGN_Q.get("avatarsQ", ""), {"id": id}, "Avatars" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.CAMPAIGN_Q.get("avatarsQ", ""), {"id": id}, "Avatars").get("data", {}) def getFollowRequestWatchCount(self) -> dict[str, Any]: """Get the follow request and watch count data. @@ -869,9 +815,7 @@ def ranks(self, campaignId: str) -> dict[str, Any]: Returns: dict[str, Any]: The data returned by the query, in the form of a dictionary. """ - return self.runAuthorizedGqlQuery( - gq.CAMPAIGN_Q.get("ranksQ", ""), {"campaignId": campaignId}, "Ranks" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.CAMPAIGN_Q.get("ranksQ", ""), {"campaignId": campaignId}, "Ranks").get("data", {}) def conv360IDToO2OID(self, qid: str, deviceId: str) -> dict[str, Any]: """Convert the 360 ID to the O2O ID. @@ -895,9 +839,7 @@ def getAppVersion(self) -> dict[str, Any]: Returns: dict[str, Any]: The data for the GetAppVersion query. """ - return self.runAuthorizedGqlQuery( - gq.QUERY.get("getAppVersionQ", ""), {}, "GetAppVersion" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.QUERY.get("getAppVersionQ", ""), {}, "GetAppVersion").get("data", {}) def watchGroups(self, id: str = "") -> dict[str, Any]: """Returns the data for the WatchGroups query. @@ -908,9 +850,7 @@ def watchGroups(self, id: str = "") -> dict[str, Any]: Returns: dict[str, Any]: The data for the WatchGroups query. """ - return self.runAuthorizedGqlQuery( - gq.WATCHGROUP_Q.get("watchGroupsQ", ""), {"id": id}, "WatchGroups" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gq.WATCHGROUP_Q.get("watchGroupsQ", ""), {"id": id}, "WatchGroups").get("data", {}) def getStartTrackingWatch(self, wuid: str) -> dict[str, Any]: """Returns the data for the StartTrackingWatch query. @@ -940,9 +880,7 @@ def getEndTrackingWatch(self, wuid: str) -> dict[str, Any]: Returns: dict[str, Any]: The data for the EndTrackingWatch query. """ - data = self.runAuthorizedGqlQuery( - gq.WATCH_Q.get("endTrackingWatchQ", ""), {"uid": wuid}, "EndTrackingWatch" - ) + data = self.runAuthorizedGqlQuery(gq.WATCH_Q.get("endTrackingWatchQ", ""), {"uid": wuid}, "EndTrackingWatch") errors: list[dict[str, str]] = data.get("errors", []) if errors: self.errors.append({"function": "getEndTrackingWatch", "error": errors}) @@ -997,7 +935,9 @@ def sendText(self, wuid: str, text: str) -> bool: gm.WATCH_M.get("sendChatTextM", ""), {"uid": wuid, "text": text}, "SendChatText", - ).get("data", {})["sendChatText"] + ).get( + "data", {} + )["sendChatText"] is not None ): return True @@ -1012,9 +952,7 @@ def addStep(self, stepCount: int) -> dict[str, Any]: Returns: dict: A dictionary containing the data from the response. """ - return self.runAuthorizedGqlQuery( - gm.STEP_M.get("addM", ""), {"stepCount": stepCount}, "AddStep" - ).get("data", {}) + return self.runAuthorizedGqlQuery(gm.STEP_M.get("addM", ""), {"stepCount": stepCount}, "AddStep").get("data", {}) def shutdown(self, wuid: str) -> bool: """Shuts down the system for a specified user. @@ -1025,9 +963,7 @@ def shutdown(self, wuid: str) -> bool: Returns: bool: Whether the system was shut down successfully. """ - return self.isAdmin( - wuid, gm.WATCH_M.get("shutdownM", ""), {"uid": wuid}, "ShutDown" - ) + return self.isAdmin(wuid, gm.WATCH_M.get("shutdownM", ""), {"uid": wuid}, "ShutDown") def reboot(self, wuid: str) -> bool: """Reboots the system for a specified user. @@ -1038,9 +974,7 @@ def reboot(self, wuid: str) -> bool: Returns: bool: Whether the system was rebooted successfully. """ - return self.isAdmin( - wuid, gm.WATCH_M.get("rebootM", ""), {"uid": wuid}, "reboot" - ) + return self.isAdmin(wuid, gm.WATCH_M.get("rebootM", ""), {"uid": wuid}, "reboot") def modifyAlert(self, id: str, yesOrNo: str) -> dict[str, Any]: """Modifies an alert. @@ -1058,9 +992,7 @@ def modifyAlert(self, id: str, yesOrNo: str) -> dict[str, Any]: "modifyAlert", ) - def setEnableSilentTime( - self, silent_id: str, status: str = NormalStatus.ENABLE.value - ) -> dict[str, Any]: + def setEnableSilentTime(self, silent_id: str, status: str = NormalStatus.ENABLE.value) -> dict[str, Any]: """Sets the silent time for a specified user. Args: @@ -1076,9 +1008,7 @@ def setEnableSilentTime( "SetEnableSlientTime", ).get("data", {}) - def setEnableAlarmTime( - self, alarm_id: str, status: str = NormalStatus.ENABLE.value - ) -> dict[str, Any]: + def setEnableAlarmTime(self, alarm_id: str, status: str = NormalStatus.ENABLE.value) -> dict[str, Any]: """Enable or disable alarm time. Args: @@ -1113,9 +1043,7 @@ def setReadChatMsg(self, wuid: str, msgId: str, id: str) -> dict[str, Any]: "setReadChatMsg", ).get("data", {}) - def submitIncorrectLocationData( - self, wuid: str, lat: str, lng: str, timestamp: str - ) -> dict[str, Any]: + def submitIncorrectLocationData(self, wuid: str, lat: str, lng: str, timestamp: str) -> dict[str, Any]: """Submit incorrect location data. Args: @@ -1133,9 +1061,7 @@ def submitIncorrectLocationData( "SubmitIncorrectLocationData", ).get("data", {}) - def modifyContact( - self, contactId: str, isAdmin: bool, contactName: str = "", fileId: str = "" - ) -> dict[str, Any]: + def modifyContact(self, contactId: str, isAdmin: bool, contactName: str = "", fileId: str = "") -> dict[str, Any]: """Modify a contact. Args: diff --git a/src/pyxplora_api/gql_handler_async.py b/src/pyxplora_api/gql_handler_async.py index a24287e..5c59eb6 100644 --- a/src/pyxplora_api/gql_handler_async.py +++ b/src/pyxplora_api/gql_handler_async.py @@ -32,9 +32,7 @@ def __init__( ) -> None: self._session = session self.refreshToken = None - super().__init__( - countryPhoneNumber, phoneNumber, password, userLang, timeZone, email, signup - ) + super().__init__(countryPhoneNumber, phoneNumber, password, userLang, timeZone, email, signup) async def runGqlQuery_a( self, @@ -88,14 +86,10 @@ async def login_a(self, key, sec) -> tuple[dict[str, Any], Any]: data = dataAll.get("data", {}) signIn: dict[str, Any] | None = data.get("signInWithEmailOrPhone", None) if signIn is None: - error_message = dataAll.get("errors", [{"message": ""}])[0].get( - "message", "" - ) + error_message = dataAll.get("errors", [{"message": ""}])[0].get("message", "") if error_message: raise LoginError(f"Login error: {error_message}") - raise LoginError( - "The server is not responding, please wait a moment and try again." - ) + raise LoginError("The server is not responding, please wait a moment and try again.") self.issueToken = signIn self.refreshToken = self.issueToken.get("refreshToken", None) @@ -110,9 +104,7 @@ async def login_a(self, key, sec) -> tuple[dict[str, Any], Any]: return self.issueToken, self.refreshToken - async def isAdmin_a( - self, wuid: str, query: str, variables: dict[str, Any], key: str - ) -> bool: + async def isAdmin_a(self, wuid: str, query: str, variables: dict[str, Any], key: str) -> bool: contacts: dict[str, Any] = await self.getWatchUserContacts_a(wuid) for contact in contacts["contacts"]["contacts"]: try: @@ -121,9 +113,7 @@ async def isAdmin_a( id = None if self.userId == id: if contact["guardianType"] == "FIRST": - data: dict[str, Any] = ( - await self.runGqlQuery_a(query, variables, key) - ).get("data", {}) + data: dict[str, Any] = (await self.runGqlQuery_a(query, variables, key)).get("data", {}) for k in data: if k.upper() == key.upper(): return data.get(k, False) @@ -132,9 +122,7 @@ async def isAdmin_a( ########## SECTION QUERY start ########## async def askWatchLocate_a(self, wuid: str) -> dict[str, Any]: - data: dict[str, Any] = await self.runGqlQuery_a( - gq.WATCH_Q.get("askLocateQ", ""), {"uid": wuid}, "AskWatchLocate" - ) + data: dict[str, Any] = await self.runGqlQuery_a(gq.WATCH_Q.get("askLocateQ", ""), {"uid": wuid}, "AskWatchLocate") errors = data.get("errors", []) if errors: self.errors.append({"function": "askWatchLocate", "errors": errors}) @@ -145,18 +133,14 @@ async def askWatchLocate_a(self, wuid: str) -> dict[str, Any]: async def getWatchUserContacts_a(self, wuid: str) -> dict[str, Any]: # Contacts from ownUser - data: dict[str, Any] = await self.runGqlQuery_a( - gq.WATCH_Q.get("contactsQ", ""), {"uid": wuid}, "Contacts" - ) + data: dict[str, Any] = await self.runGqlQuery_a(gq.WATCH_Q.get("contactsQ", ""), {"uid": wuid}, "Contacts") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatchUserContacts", "errors": errors}) return data.get("data", {}) async def getWatches_a(self, wuid: str) -> dict[str, Any]: - data: dict[str, Any] = await self.runGqlQuery_a( - gq.WATCH_Q.get("watchesQ", ""), {"uid": wuid}, "Watches" - ) + data: dict[str, Any] = await self.runGqlQuery_a(gq.WATCH_Q.get("watchesQ", ""), {"uid": wuid}, "Watches") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatches", "errors": errors}) @@ -173,9 +157,7 @@ async def getSWInfo_a(self, qrCode: str) -> dict[str, Any]: self.errors.append({"function": "getSWInfo", "errors": errors}) return data.get("data", {}) - async def getWatchState_a( - self, qrCode: str, qrt: str = "", qrc: str = "" - ) -> dict[str, Any]: + async def getWatchState_a(self, qrCode: str, qrt: str = "", qrc: str = "") -> dict[str, Any]: variables = {} if qrCode: variables["qrCode"] = qrCode @@ -183,24 +165,18 @@ async def getWatchState_a( variables["qrt"] = qrt if qrc: variables["qrc"] = qrc - data: dict[str, Any] = await self.runGqlQuery_a( - gq.WATCH_Q.get("stateQ", ""), variables, "WatchState" - ) + data: dict[str, Any] = await self.runGqlQuery_a(gq.WATCH_Q.get("stateQ", ""), variables, "WatchState") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatchState", "errors": errors}) return data.get("data", {}) async def getWatchLastLocation_a(self, wuid: str) -> dict[str, Any]: - data: dict[str, Any] = await self.runGqlQuery_a( - gq.WATCH_Q.get("locateQ", ""), {"uid": wuid}, "WatchLastLocate" - ) + data: dict[str, Any] = await self.runGqlQuery_a(gq.WATCH_Q.get("locateQ", ""), {"uid": wuid}, "WatchLastLocate") errors = data.get("errors", []) if errors: self.errors.append({"function": "getWatchLastLocation", "errors": errors}) - error_msg = data.get("errors", [{"code": "E", "message": "E"}])[0].get( - "message", "E" - ) + error_msg = data.get("errors", [{"code": "E", "message": "E"}])[0].get("message", "E") if error_msg == ErrorMSG.AUTH_FAIL.value: _LOGGER.error(error_msg) return errors[0] @@ -208,9 +184,7 @@ async def getWatchLastLocation_a(self, wuid: str) -> dict[str, Any]: async def trackWatch_a(self, wuid: str) -> dict[str, Any]: # tracking time - seconds - data: dict[str, Any] = await self.runGqlQuery_a( - gq.WATCH_Q.get("trackQ", ""), {"uid": wuid}, "TrackWatch" - ) + data: dict[str, Any] = await self.runGqlQuery_a(gq.WATCH_Q.get("trackQ", ""), {"uid": wuid}, "TrackWatch") errors = data.get("errors", []) if errors: self.errors.append({"function": "trackWatch", "errors": errors}) @@ -220,19 +194,11 @@ async def trackWatch_a(self, wuid: str) -> dict[str, Any]: return {"trackWatch": -1} async def getAlarmTime_a(self, wuid: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.WATCH_Q.get("alarmsQ", ""), {"uid": wuid}, "Alarms" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCH_Q.get("alarmsQ", ""), {"uid": wuid}, "Alarms")).get("data", {}) async def getWifi_a(self, wuid: str) -> dict[str, Any]: # without function? - return ( - await self.runGqlQuery_a( - gq.WATCH_Q.get("getWifisQ", ""), {"uid": wuid}, "GetWifis" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCH_Q.get("getWifisQ", ""), {"uid": wuid}, "GetWifis")).get("data", {}) async def unReadChatMsgCount_a(self, wuid: str) -> dict[str, Any]: return ( @@ -244,25 +210,13 @@ async def unReadChatMsgCount_a(self, wuid: str) -> dict[str, Any]: ).get("data", {}) async def safeZones_a(self, wuid: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.WATCH_Q.get("safeZonesQ", ""), {"uid": wuid}, "SafeZones" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCH_Q.get("safeZonesQ", ""), {"uid": wuid}, "SafeZones")).get("data", {}) async def safeZoneGroups_a(self) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.WATCH_Q.get("safeZoneGroupsQ", ""), {}, "SafeZoneGroups" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCH_Q.get("safeZoneGroupsQ", ""), {}, "SafeZoneGroups")).get("data", {}) async def silentTimes_a(self, wuid: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.WATCH_Q.get("silentTimesQ", ""), {"uid": wuid}, "SlientTimes" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCH_Q.get("silentTimesQ", ""), {"uid": wuid}, "SlientTimes")).get("data", {}) async def chats_a( self, @@ -332,9 +286,7 @@ async def fetchChatVoice_a(self, wuid: str, msgId: str) -> dict[str, Any]: ) ).get("data", {}) - async def watchImei_a( - self, imei: str, qrCode: str, deviceKey: str - ) -> dict[str, Any]: + async def watchImei_a(self, imei: str, qrCode: str, deviceKey: str) -> dict[str, Any]: return ( await self.runGqlQuery_a( gq.WATCH_Q.get("imeiQ", ""), @@ -355,15 +307,9 @@ async def getWatchLocHistory_a( ).get("data", {}) async def watchesDynamic_a(self) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.WATCH_Q.get("watchesDynamicQ", ""), {}, "WatchesDynamic" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCH_Q.get("watchesDynamicQ", ""), {}, "WatchesDynamic")).get("data", {}) - async def coinHistory_a( - self, wuid: str, start: int, end: int, _type: str, offset: int, limit: int - ) -> dict[str, Any]: + async def coinHistory_a(self, wuid: str, start: int, end: int, _type: str, offset: int, limit: int) -> dict[str, Any]: return ( await self.runGqlQuery_a( gq.XCOIN_Q.get("historyQ", ""), @@ -380,32 +326,20 @@ async def coinHistory_a( ).get("data", {}) async def reminders_a(self, wuid: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.XMOVE_Q.get("remindersQ", ""), {"uid": wuid}, "Reminders" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.XMOVE_Q.get("remindersQ", ""), {"uid": wuid}, "Reminders")).get("data", {}) async def groups_a(self, isCampaign: bool) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.CARD_Q.get("groupsQ", ""), {"isCampaign": isCampaign}, "CardGroups" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.CARD_Q.get("groupsQ", ""), {"isCampaign": isCampaign}, "CardGroups")).get( + "data", {} + ) async def dynamic_a(self) -> dict[str, Any]: - return ( - await self.runGqlQuery_a(gq.CARD_Q.get("dynamicQ", ""), {}, "DynamicCards") - ).get("data", {}) + return (await self.runGqlQuery_a(gq.CARD_Q.get("dynamicQ", ""), {}, "DynamicCards")).get("data", {}) async def staticCard_a(self) -> dict[str, Any]: - return ( - await self.runGqlQuery_a(gq.CARD_Q.get("staticQ", ""), {}, "StaticCard") - ).get("data", {}) + return (await self.runGqlQuery_a(gq.CARD_Q.get("staticQ", ""), {}, "StaticCard")).get("data", {}) - async def familyInfo_a( - self, wuid: str, watchId: str, tz: str, date: int - ) -> dict[str, Any]: + async def familyInfo_a(self, wuid: str, watchId: str, tz: str, date: int) -> dict[str, Any]: return ( await self.runGqlQuery_a( gq.FAMILY_Q.get("infoQ", ""), @@ -470,9 +404,7 @@ async def myInfoWithCoinHistory_a( async def getMyInfo_a(self) -> dict[str, Any]: # Profil from login Account - return ( - await self.runGqlQuery_a(gq.MYINFO_Q.get("readQ", ""), {}, "ReadMyInfo") - ).get("data", {}) + return (await self.runGqlQuery_a(gq.MYINFO_Q.get("readQ", ""), {}, "ReadMyInfo")).get("data", {}) async def readCampaignProfile_a(self, wuid: str) -> dict[str, Any]: return ( @@ -483,15 +415,9 @@ async def readCampaignProfile_a(self, wuid: str) -> dict[str, Any]: ).get("data", {}) async def getReviewStatus_a(self, wuid: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.REVIEW_Q.get("getStatusQ", ""), {"uid": wuid}, "GetReviewStatus" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.REVIEW_Q.get("getStatusQ", ""), {"uid": wuid}, "GetReviewStatus")).get("data", {}) - async def getWatchUserSteps_a( - self, wuid: str, tz: str, date: int - ) -> dict[str, Any]: + async def getWatchUserSteps_a(self, wuid: str, tz: str, date: int) -> dict[str, Any]: data: dict[str, Any] = await self.runGqlQuery_a( gq.STEP_Q.get("userQ", ""), {"uid": wuid, "tz": tz, "date": date}, @@ -504,16 +430,10 @@ async def getWatchUserSteps_a( async def countries_a(self) -> dict[str, Any]: # Country Support - return ( - await self.runGqlQuery_a(gq.UTILS_Q.get("countriesQ", ""), {}, "Countries") - ).get("data", {}) + return (await self.runGqlQuery_a(gq.UTILS_Q.get("countriesQ", ""), {}, "Countries")).get("data", {}) async def avatars_a(self, _id: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.CAMPAIGN_Q.get("avatarsQ", ""), {"id": _id}, "Avatars" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.CAMPAIGN_Q.get("avatarsQ", ""), {"id": _id}, "Avatars")).get("data", {}) async def getFollowRequestWatchCount_a(self) -> dict[str, Any]: return ( @@ -552,11 +472,7 @@ async def subscribed_a(self, wuid: str, needDetail: bool) -> dict[str, Any]: ).get("data", {}) async def ranks_a(self, campaignId: str) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.CAMPAIGN_Q.get("ranksQ", ""), {"campaignId": campaignId}, "Ranks" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.CAMPAIGN_Q.get("ranksQ", ""), {"campaignId": campaignId}, "Ranks")).get("data", {}) async def conv360IDToO2OID_a(self, qid: str, deviceId: str) -> dict[str, Any]: return ( @@ -568,18 +484,10 @@ async def conv360IDToO2OID_a(self, qid: str, deviceId: str) -> dict[str, Any]: ).get("data", {}) async def getAppVersion_a(self) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.QUERY.get("getAppVersionQ", ""), {}, "GetAppVersion" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.QUERY.get("getAppVersionQ", ""), {}, "GetAppVersion")).get("data", {}) async def watchGroups_a(self, _id: str = "") -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gq.WATCHGROUP_Q.get("watchGroupsQ", ""), {"id": _id}, "WatchGroups" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gq.WATCHGROUP_Q.get("watchGroupsQ", ""), {"id": _id}, "WatchGroups")).get("data", {}) async def getStartTrackingWatch_a(self, wuid: str) -> dict[str, Any]: data = await self.runGqlQuery_a( @@ -593,9 +501,7 @@ async def getStartTrackingWatch_a(self, wuid: str) -> dict[str, Any]: return data.get("data", {}) async def getEndTrackingWatch_a(self, wuid: str) -> dict[str, Any]: - data = await self.runGqlQuery_a( - gq.WATCH_Q.get("endTrackingWatchQ", ""), {"uid": wuid}, "EndTrackingWatch" - ) + data = await self.runGqlQuery_a(gq.WATCH_Q.get("endTrackingWatchQ", ""), {"uid": wuid}, "EndTrackingWatch") errors: list[dict[str, str]] = data.get("errors", []) if errors: self.errors.append({"function": "getEndTrackingWatch", "error": errors}) @@ -640,23 +546,15 @@ async def sendText_a(self, wuid: str, text: str) -> bool: return False async def addStep_a(self, stepCount: int) -> dict[str, Any]: - return ( - await self.runGqlQuery_a( - gm.STEP_M.get("addM", ""), {"stepCount": stepCount}, "AddStep" - ) - ).get("data", {}) + return (await self.runGqlQuery_a(gm.STEP_M.get("addM", ""), {"stepCount": stepCount}, "AddStep")).get("data", {}) async def shutdown_a(self, wuid: str) -> bool: # ownUser id - return await self.isAdmin_a( - wuid, gm.WATCH_M.get("shutdownM", ""), {"uid": wuid}, "ShutDown" - ) + return await self.isAdmin_a(wuid, gm.WATCH_M.get("shutdownM", ""), {"uid": wuid}, "ShutDown") async def reboot_a(self, wuid: str) -> bool: # ownUser id - return await self.isAdmin_a( - wuid, gm.WATCH_M.get("rebootM", ""), {"uid": wuid}, "reboot" - ) + return await self.isAdmin_a(wuid, gm.WATCH_M.get("rebootM", ""), {"uid": wuid}, "reboot") async def modifyAlert_a(self, _id: str, yesOrNo: str) -> dict[str, Any]: # function? @@ -666,9 +564,7 @@ async def modifyAlert_a(self, _id: str, yesOrNo: str) -> dict[str, Any]: "modifyAlert", ) - async def setEnableSilentTime_a( - self, silent_id: str, status: str = NormalStatus.ENABLE.value - ) -> dict[str, Any]: + async def setEnableSilentTime_a(self, silent_id: str, status: str = NormalStatus.ENABLE.value) -> dict[str, Any]: return ( await self.runGqlQuery_a( gm.WATCH_M.get("setEnableSlientTimeM", ""), @@ -677,9 +573,7 @@ async def setEnableSilentTime_a( ) ).get("data", {}) - async def setEnableAlarmTime_a( - self, alarm_id: str, status: str = NormalStatus.ENABLE.value - ) -> dict[str, Any]: + async def setEnableAlarmTime_a(self, alarm_id: str, status: str = NormalStatus.ENABLE.value) -> dict[str, Any]: return ( await self.runGqlQuery_a( gm.WATCH_M.get("modifyAlarmM", ""), @@ -697,9 +591,7 @@ async def setReadChatMsg_a(self, wuid: str, msgId: str, _id: str) -> dict[str, A ) ).get("data", {}) - async def submitIncorrectLocationData_a( - self, wuid: str, lat: str, lng: str, timestamp: str - ) -> dict[str, Any]: + async def submitIncorrectLocationData_a(self, wuid: str, lat: str, lng: str, timestamp: str) -> dict[str, Any]: return ( await self.runGqlQuery_a( gm.WATCH_M.get("submitIncorrectLocationDataM", ""), @@ -708,9 +600,7 @@ async def submitIncorrectLocationData_a( ) ).get("data", {}) - async def modifyContact_a( - self, contactId: str, isAdmin: bool, contactName: str = "", fileId: str = "" - ) -> dict[str, Any]: + async def modifyContact_a(self, contactId: str, isAdmin: bool, contactName: str = "", fileId: str = "") -> dict[str, Any]: return await self.runGqlQuery_a( gm.WATCH_M.get("modifyContactM", ""), { @@ -767,9 +657,7 @@ async def signUpWithEmailAndPhoneV2_a( "SignUpWithEmailAndPhoneV2", ) - async def verifyCaptcha_a( - self, captchaString: str = "", _type: str = "" - ) -> dict[str, Any]: + async def verifyCaptcha_a(self, captchaString: str = "", _type: str = "") -> dict[str, Any]: return await self.runGqlQuery_a( gm.SIGN_M.get("verifyCaptchaM", ""), {"captchaString": captchaString, "type": _type}, @@ -808,9 +696,7 @@ async def deleteMessageFromApp_a(self, wuid: str, msgId: str): ).get("data", {}) async def connect360_a(self): - data = await self.runGqlQuery_a( - gm.SIGN_M.get("connect360M", ""), {}, "connect360" - ) + data = await self.runGqlQuery_a(gm.SIGN_M.get("connect360M", ""), {}, "connect360") return data.get("data", {}) async def refresh_token_a(self, wuid: str, refresh_token: str) -> str | None: diff --git a/src/pyxplora_api/gql_queries.py b/src/pyxplora_api/gql_queries.py index 9316d16..229b091 100644 --- a/src/pyxplora_api/gql_queries.py +++ b/src/pyxplora_api/gql_queries.py @@ -31,9 +31,7 @@ "readCampaignProfileQ": "query ReadMyInfoWithCampaignProfile {\n campaignUserProfiles {\n __typename\n ...CampaignUserProfilesFragment\n }\n readMyInfo {\n __typename\n ...UserFragment\n }\n followRequestWatchCount\n}\nfragment CampaignUserProfilesFragment on CampaignUserProfile {\n __typename\n id\n user {\n __typename\n ...UserFragment\n }\n name\n avatar {\n __typename\n ...AvatarFragment\n }\n profile {\n __typename\n ...FileFragment\n }\n create\n update\n}\nfragment UserFragment on User {\n __typename\n id\n userId\n name\n nickname\n gender\n birth\n birthStr\n weight\n height\n countryCode\n emailAddress\n emailConsent\n countryPhoneCode\n phoneNumber\n mobilePhoneNumber\n emailConfirm\n status\n file {\n __typename\n ...FileFragment\n }\n extra\n xcoin\n currentStep\n totalStep\n create\n update\n children {\n __typename\n id\n guardian {\n __typename\n ...SimpleUserFragment\n }\n ward {\n __typename\n ...SimpleUserFragment\n }\n }\n}\nfragment FileFragment on File {\n __typename\n id\n name\n}\nfragment SimpleUserFragment on User {\n __typename\n id\n userId\n name\n nickname\n gender\n countryCode\n countryPhoneCode\n phoneNumber\n mobilePhoneNumber\n file {\n __typename\n ...FileFragment\n }\n xcoin\n currentStep\n totalStep\n extra\n contacts {\n __typename\n ...ContactsFragment\n }\n}\nfragment ContactsFragment on Contact {\n __typename\n id\n me {\n __typename\n ...ContactorFragment\n }\n contacter {\n __typename\n ...ContactorFragment\n }\n phoneNumber\n extra\n listOrder\n file {\n __typename\n ...FileFragment\n }\n create\n update\n}\nfragment ContactorFragment on User {\n __typename\n id\n userId\n name\n nickname\n countryCode\n countryPhoneCode\n mobilePhoneNumber\n phoneNumber\n}\nfragment AvatarFragment on Avatar {\n __typename\n id\n name\n file {\n __typename\n ...FileFragment\n }\n}", "campaignUserProfilesQ": "query CampaignUserProfiles {\n campaignUserProfiles {\n __typename\n ...CampaignUserProfilesFragment\n }\n followRequestWatchCount\n}\nfragment CampaignUserProfilesFragment on CampaignUserProfile {\n __typename\n id\n user {\n __typename\n ...UserFragment\n }\n name\n avatar {\n __typename\n ...AvatarFragment\n }\n profile {\n __typename\n ...FileFragment\n }\n create\n update\n}\nfragment UserFragment on User {\n __typename\n id\n userId\n name\n nickname\n gender\n birth\n birthStr\n weight\n height\n countryCode\n emailAddress\n emailConsent\n countryPhoneCode\n phoneNumber\n mobilePhoneNumber\n emailConfirm\n status\n file {\n __typename\n ...FileFragment\n }\n extra\n xcoin\n currentStep\n totalStep\n create\n update\n children {\n __typename\n id\n guardian {\n __typename\n ...SimpleUserFragment\n }\n ward {\n __typename\n ...SimpleUserFragment\n }\n }\n}\nfragment FileFragment on File {\n __typename\n id\n name\n}\nfragment SimpleUserFragment on User {\n __typename\n id\n userId\n name\n nickname\n gender\n countryCode\n countryPhoneCode\n phoneNumber\n mobilePhoneNumber\n file {\n __typename\n ...FileFragment\n }\n xcoin\n currentStep\n totalStep\n extra\n contacts {\n __typename\n ...ContactsFragment\n }\n}\nfragment ContactsFragment on Contact {\n __typename\n id\n me {\n __typename\n ...ContactorFragment\n }\n contacter {\n __typename\n ...ContactorFragment\n }\n phoneNumber\n extra\n listOrder\n file {\n __typename\n ...FileFragment\n }\n create\n update\n}\nfragment ContactorFragment on User {\n __typename\n id\n userId\n name\n nickname\n countryCode\n countryPhoneCode\n mobilePhoneNumber\n phoneNumber\n}\nfragment AvatarFragment on Avatar {\n __typename\n id\n name\n file {\n __typename\n ...FileFragment\n }\n}", } -REVIEW_Q: dict[str, str] = { - "getStatusQ": "query GetReviewStatus($uid: String!) {\n getReviewStatus(uid: $uid)\n}" -} +REVIEW_Q: dict[str, str] = {"getStatusQ": "query GetReviewStatus($uid: String!) {\n getReviewStatus(uid: $uid)\n}"} STEP_Q: dict[str, str] = { "userQ": "query UserSteps($uid: String!, $tz: String, $date : Int!) {\n userSteps(uid: $uid, tz : $tz, date: $date) {\n __typename\n ...UserStepsFragment\n }\n}\nfragment UserStepsFragment on UserStep {\n __typename\n age\n user {\n __typename\n ...UserFragment\n }\n day\n monthSteps {\n __typename\n ...StepItemFragment\n }\n daySteps {\n __typename\n ...StepItemFragment\n }\n timeSteps {\n __typename\n ...StepItemFragment\n }\n}\nfragment UserFragment on User {\n __typename\n id\n userId\n name\n nickname\n gender\n birth\n birthStr\n weight\n height\n countryCode\n emailAddress\n emailConsent\n countryPhoneCode\n phoneNumber\n mobilePhoneNumber\n emailConfirm\n status\n file {\n __typename\n ...FileFragment\n }\n extra\n xcoin\n currentStep\n totalStep\n create\n update\n children {\n __typename\n id\n guardian {\n __typename\n ...SimpleUserFragment\n }\n ward {\n __typename\n ...SimpleUserFragment\n }\n }\n}\nfragment FileFragment on File {\n __typename\n id\n name\n}\nfragment SimpleUserFragment on User {\n __typename\n id\n userId\n name\n nickname\n gender\n countryCode\n countryPhoneCode\n phoneNumber\n mobilePhoneNumber\n file {\n __typename\n ...FileFragment\n }\n xcoin\n currentStep\n totalStep\n extra\n contacts {\n __typename\n ...ContactsFragment\n }\n}\nfragment ContactsFragment on Contact {\n __typename\n id\n me {\n __typename\n ...ContactorFragment\n }\n contacter {\n __typename\n ...ContactorFragment\n }\n phoneNumber\n extra\n listOrder\n file {\n __typename\n ...FileFragment\n }\n create\n update\n}\nfragment ContactorFragment on User {\n __typename\n id\n userId\n name\n nickname\n countryCode\n countryPhoneCode\n mobilePhoneNumber\n phoneNumber\n}\nfragment StepItemFragment on StepItem {\n __typename\n key\n step\n}", } diff --git a/src/pyxplora_api/graphql_client.py b/src/pyxplora_api/graphql_client.py index b4b7891..6b1f255 100644 --- a/src/pyxplora_api/graphql_client.py +++ b/src/pyxplora_api/graphql_client.py @@ -14,9 +14,7 @@ class GraphqlClient: """Class which represents the interface to make graphQL requests through.""" - def __init__( - self, endpoint: str, headers: Optional[dict[str, str]] = None, **kwargs: Any - ): + def __init__(self, endpoint: str, headers: Optional[dict[str, str]] = None, **kwargs: Any): """Instantiate the client.""" headers = {} if headers is None else headers self.logger = logging.getLogger(__name__) @@ -49,9 +47,7 @@ def execute( ): """Make synchronous request to graphQL server.""" headers = {} if headers is None else headers - request_body = self.__request_body( - query=query, variables=variables, operation_name=operation_name - ) + request_body = self.__request_body(query=query, variables=variables, operation_name=operation_name) if "user-agent" not in headers: headers["user-agent"] = DEFAULT_USER_AGENT @@ -75,15 +71,11 @@ async def execute_async( ): """Make asynchronous request to graphQL server.""" headers = {} if headers is None else headers - request_body = self.__request_body( - query=query, variables=variables, operation_name=operation_name - ) + request_body = self.__request_body(query=query, variables=variables, operation_name=operation_name) if "user-agent" not in headers: headers["user-agent"] = DEFAULT_USER_AGENT - async with aiohttp.ClientSession( - timeout=aiohttp.ClientTimeout(DEFAULT_TIMEOUT) - ) as session, session.post( + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(DEFAULT_TIMEOUT)) as session, session.post( self.endpoint, json=request_body, headers={**self.headers, **headers} ) as response: try: @@ -103,9 +95,7 @@ async def ha_execute_async( ): """Make asynchronous request to graphQL server.""" headers = {} if headers is None else headers - request_body = self.__request_body( - query=query, variables=variables, operation_name=operation_name - ) + request_body = self.__request_body(query=query, variables=variables, operation_name=operation_name) if "user-agent" not in headers: headers["user-agent"] = DEFAULT_USER_AGENT @@ -116,9 +106,7 @@ async def ha_execute_async( operation_name=operation_name, headers=headers, ) - async with session.post( - self.endpoint, json=request_body, headers={**self.headers, **headers} - ) as response: + async with session.post(self.endpoint, json=request_body, headers={**self.headers, **headers}) as response: try: response.raise_for_status() return await response.json() diff --git a/src/pyxplora_api/handler_gql.py b/src/pyxplora_api/handler_gql.py index f2fec5d..c5eb36c 100644 --- a/src/pyxplora_api/handler_gql.py +++ b/src/pyxplora_api/handler_gql.py @@ -114,9 +114,7 @@ def getRequestHeaders(self, acceptedContentType: str) -> dict[str, Any]: authorizationHeader = "" - if ( - self.accessToken is None or not self.issueToken - ) and self._API_KEY == API_KEY: + if (self.accessToken is None or not self.issueToken) and self._API_KEY == API_KEY: # OPEN authorization authorizationHeader = f"Open {self._API_KEY}:{self._API_SECRET}" # else: @@ -125,7 +123,9 @@ def getRequestHeaders(self, acceptedContentType: str) -> dict[str, Any]: w360: dict = self.issueToken.get("w360", None) if w360: if w360.get("token") and w360.get("secret"): - authorizationHeader = f'Bearer {w360.get("token", self.accessToken)}:{w360.get("secret", self._API_SECRET)}' + authorizationHeader = ( + f'Bearer {w360.get("token", self.accessToken)}:{w360.get("secret", self._API_SECRET)}' + ) self._API_KEY = w360.get("token", API_KEY) self._API_SECRET = w360.get("secret", API_SECRET) else: diff --git a/src/pyxplora_api/model.py b/src/pyxplora_api/model.py index b8dd673..c014997 100644 --- a/src/pyxplora_api/model.py +++ b/src/pyxplora_api/model.py @@ -54,9 +54,7 @@ class Data(DataClassJsonMixin): sender_name: Union[str, None] = None text: Union[str, None] = None Text: Union[str, None] = None - battery: Union[int, None] = field( - default=None, metadata=config(decoder=int_or_none) - ) + battery: Union[int, None] = field(default=None, metadata=config(decoder=int_or_none)) poi: Union[str, None] = None city: Union[str, None] = None address: Union[str, None] = None @@ -65,12 +63,8 @@ class Data(DataClassJsonMixin): emoticon_id: Union[Emoticon.__str__, None] = Emoticon.UNKNOWN__.value emoji_id: Union[Emoticon.__str__, None] = Emoticon.UNKNOWN__.value call_name: Union[str, None] = None - call_time: Union[int, None] = field( - default=None, metadata=config(decoder=int_or_none) - ) - call_type: Union[int, None] = field( - default=None, metadata=config(decoder=int_or_none) - ) + call_time: Union[int, None] = field(default=None, metadata=config(decoder=int_or_none)) + call_type: Union[int, None] = field(default=None, metadata=config(decoder=int_or_none)) lat: Union[float, None] = None lng: Union[float, None] = None radius: Union[int, None] = field(default=None, metadata=config(decoder=int_or_none)) @@ -82,9 +76,7 @@ class Data(DataClassJsonMixin): class SimpleChat(DataClassJsonMixin): id: Union[str, None] = None msgId: Union[str, None] = None # noqa: N815 - readFlag: Union[int, None] = field( - default=None, metadata=config(decoder=int_or_none) - ) # noqa: N815 + readFlag: Union[int, None] = field(default=None, metadata=config(decoder=int_or_none)) # noqa: N815 sender: Union[User, None] = None receiver: Union[User, None] = None data: Union[Data, None] = None diff --git a/src/pyxplora_api/pyxplora.py b/src/pyxplora_api/pyxplora.py index 7073e0e..35023cd 100644 --- a/src/pyxplora_api/pyxplora.py +++ b/src/pyxplora_api/pyxplora.py @@ -75,9 +75,7 @@ def _isConnected(self) -> bool: Returns: bool: True if the instance is connected, False otherwise. """ - return bool( - self._gql_handler and self._issueToken and self._gql_handler.accessToken - ) + return bool(self._gql_handler and self._issueToken and self._gql_handler.accessToken) def _logoff(self) -> None: """Log off the user by clearing the stored information. @@ -183,9 +181,7 @@ def getUserCreate(self) -> str: Returns: str: The creation date and time of the user in the format "YYYY-MM-DD HH:MM:SS". """ - return datetime.fromtimestamp(self.user.get("create", 0.0)).strftime( - "%Y-%m-%d %H:%M:%S" - ) + return datetime.fromtimestamp(self.user.get("create", 0.0)).strftime("%Y-%m-%d %H:%M:%S") def getUserUpdate(self) -> str: """This function returns the user update time in a string format. @@ -193,14 +189,10 @@ def getUserUpdate(self) -> str: Returns: str: The user update time in the format 'YYYY-MM-DD HH:MM:SS'. """ - return datetime.fromtimestamp(self.user.get("update", 0.0)).strftime( - "%Y-%m-%d %H:%M:%S" - ) + return datetime.fromtimestamp(self.user.get("update", 0.0)).strftime("%Y-%m-%d %H:%M:%S") ##### Watch Info ##### - def getWatchUserIDs( - self, watch_user_phone_numbers: list[str] | None = None - ) -> list[str]: + def getWatchUserIDs(self, watch_user_phone_numbers: list[str] | None = None) -> list[str]: """This function returns the unique identifiers of the watch users. Parameters: @@ -223,9 +215,7 @@ def getWatchUserIDs( watch_ids.append(watch["ward"]["id"]) return watch_ids - def getWatchUserPhoneNumbers( - self, wuid: str | list[str] | None = None, ignoreError: bool = False - ) -> str | list[str]: + def getWatchUserPhoneNumbers(self, wuid: str | list[str] | None = None, ignoreError: bool = False) -> str | list[str]: """This function returns the phone number of the watch users. Parameters: @@ -296,9 +286,7 @@ def getWatchUserNames(self, wuid: str | list[str] | None = None) -> str | list[s # raise ChildNoError(["Watch username"]) return watchusernames - def getWatchUserIcons( - self, wuid: str | list[str] | None | None = None - ) -> str | list[str]: + def getWatchUserIcons(self, wuid: str | list[str] | None | None = None) -> str | list[str]: """Get the icon URL for watch users. Parameters: @@ -320,9 +308,7 @@ def getWatchUserIcons( for watch in self.watchs: if isinstance(wuid, list): if watch["ward"]["id"] in wuid: - watch_user_icons.append( - f"https://api.myxplora.com/file?id={watch['ward']['file']['id']}" - ) + watch_user_icons.append(f"https://api.myxplora.com/file?id={watch['ward']['file']['id']}") elif isinstance(wuid, str): if watch["ward"]["id"] == wuid: return f"https://api.myxplora.com/file?id={watch['ward']['file']['id']}" @@ -332,9 +318,7 @@ def getWatchUserIcons( # raise ChildNoError(["Watch User Icon"]) return watch_user_icons - def getWatchUserXCoins( - self, wuid: str | list[str] | None = None - ) -> int | list[int]: + def getWatchUserXCoins(self, wuid: str | list[str] | None = None) -> int | list[int]: """Get the XCoins earned by the watch user. Args: @@ -366,9 +350,7 @@ def getWatchUserXCoins( # raise ChildNoError(["Watch User XCoins"]) return watchuserxcoins - def getWatchUserCurrentStep( - self, wuid: str | list[str] | None = None - ) -> int | list[int]: + def getWatchUserCurrentStep(self, wuid: str | list[str] | None = None) -> int | list[int]: """Get the current step count of a watch user. Args: @@ -399,9 +381,7 @@ def getWatchUserCurrentStep( # raise ChildNoError(["Watch User Currentsteps"]) return watchusercurrentstep - def getWatchUserTotalStep( - self, wuid: str | list[str] | None | None = None - ) -> int | list[int]: + def getWatchUserTotalStep(self, wuid: str | list[str] | None | None = None) -> int | list[int]: """Get the total steps taken by a user or a list of users from the watch. Args: diff --git a/src/pyxplora_api/pyxplora_api.py b/src/pyxplora_api/pyxplora_api.py index da1f8fd..69153fe 100644 --- a/src/pyxplora_api/pyxplora_api.py +++ b/src/pyxplora_api/pyxplora_api.py @@ -95,11 +95,7 @@ def init(self, forceLogin: bool = False, signup: bool = True) -> None: if not self._childPhoneNumber: self.watchs = children else: - self.watchs = [ - watch - for watch in children - if watch["ward"]["phoneNumber"] in self._childPhoneNumber - ] + self.watchs = [watch for watch in children if watch["ward"]["phoneNumber"] in self._childPhoneNumber] self.user = user @@ -125,31 +121,19 @@ def _setDevice(self, ids: list = None) -> list[str]: loc = self.device[wuid]["loadWatchLocation"] self.device[wuid]["watch_battery"] = int(loc.get("watch_battery", -1)) self.device[wuid]["watch_charging"] = loc.get("watch_charging", False) - self.device[wuid]["locateType"] = loc.get( - "locateType", LocationType.UNKNOWN.value - ) - self.device[wuid]["lastTrackTime"] = loc.get( - "tm", datetime.now().strftime("%Y-%m-%d %H:%M:%S") - ) + self.device[wuid]["locateType"] = loc.get("locateType", LocationType.UNKNOWN.value) + self.device[wuid]["lastTrackTime"] = loc.get("tm", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) self.device[wuid]["isInSafeZone"] = loc.get("isInSafeZone", False) self.device[wuid]["safeZoneLabel"] = loc.get("safeZoneLabel", "") self.device[wuid]["getWatchSafeZones"] = self.getWatchSafeZones(wuid=wuid) self.device[wuid]["getSilentTime"] = self.getSilentTime(wuid=wuid) self.device[wuid]["getWatches"] = self.getWatches(wuid=wuid) - self.device[wuid]["getSWInfo"] = self.getSWInfo( - wuid=wuid, watches=self.device[wuid]["getWatches"] - ) - self.device[wuid]["getWatchState"] = self.getWatchState( - wuid=wuid, watches=self.device[wuid]["getWatches"] - ) + self.device[wuid]["getSWInfo"] = self.getSWInfo(wuid=wuid, watches=self.device[wuid]["getWatches"]) + self.device[wuid]["getWatchState"] = self.getWatchState(wuid=wuid, watches=self.device[wuid]["getWatches"]) d = datetime.now() dt = datetime(year=d.year, month=d.month, day=d.day) - self.device[wuid]["getWatchUserSteps"] = self.getWatchUserSteps( - wuid=wuid, date=int(dt.timestamp()) - ) - self.device[wuid]["getWatchOnlineStatus"] = self.getWatchOnlineStatus( - wuid=wuid - ) + self.device[wuid]["getWatchUserSteps"] = self.getWatchUserSteps(wuid=wuid, date=int(dt.timestamp())) + self.device[wuid]["getWatchOnlineStatus"] = self.getWatchOnlineStatus(wuid=wuid) self.device[wuid]["getWatchUserIcons"] = self.getWatchUserIcons(wuid=wuid) self.device[wuid]["getWatchUserXCoins"] = self.getWatchUserXCoins(wuid=wuid) return wuids @@ -174,12 +158,8 @@ def getWatchUserContacts(self, wuid: str) -> list[dict[str, Any]]: { "id": id, "guardianType": contact["guardianType"], - "create": datetime.fromtimestamp( - contact["create"] - ).strftime("%Y-%m-%d %H:%M:%S"), - "update": datetime.fromtimestamp( - contact["update"] - ).strftime("%Y-%m-%d %H:%M:%S"), + "create": datetime.fromtimestamp(contact["create"]).strftime("%Y-%m-%d %H:%M:%S"), + "update": datetime.fromtimestamp(contact["update"]).strftime("%Y-%m-%d %H:%M:%S"), "name": contact["name"], "phoneNumber": f'+{contact["countryPhoneNumber"]}{contact["phoneNumber"]}', "xcoin": xcoin, @@ -219,9 +199,7 @@ def getWatchAlarm(self, wuid: str) -> list[dict[str, Any]]: self.delay(self.retryDelay) return alarms - def loadWatchLocation( - self, wuid: str = "", with_ask: bool = True - ) -> dict[str, Any]: + def loadWatchLocation(self, wuid: str = "", with_ask: bool = True) -> dict[str, Any]: retry_counter = 0 watch_location = {} while retry_counter < self.maxRetries + 1: @@ -233,11 +211,7 @@ def loadWatchLocation( if not _watch_last_locate: return watch_location - _tm = ( - 31532399 - if _watch_last_locate.get("tm") is None - else _watch_last_locate.get("tm") - ) + _tm = 31532399 if _watch_last_locate.get("tm") is None else _watch_last_locate.get("tm") _lat = _watch_last_locate.get("lat", "0.0") _lng = _watch_last_locate.get("lng", "0.0") _rad = _watch_last_locate.get("rad", -1) @@ -297,11 +271,7 @@ def getWatchOnlineStatus(self, wuid: str) -> str: try: ask_raw = self.askWatchLocate(wuid) track_raw = self.getTrackWatchInterval(wuid) - status = ( - WatchOnlineStatus.ONLINE - if ask_raw or track_raw != -1 - else WatchOnlineStatus.OFFLINE - ) + status = WatchOnlineStatus.ONLINE if ask_raw or track_raw != -1 else WatchOnlineStatus.OFFLINE except Error as error: _LOGGER.debug(error) retries += 1 @@ -335,9 +305,7 @@ def getWatchChats( while not chats and retry_counter < self.maxRetries + 2: retry_counter += 1 try: - _chats_new = self.getWatchChatsRaw( - wuid, offset, limit, msgId, show_del_msg, asObject - ) + _chats_new = self.getWatchChatsRaw(wuid, offset, limit, msgId, show_del_msg, asObject) if isinstance(_chats_new, dict): _chats_new = ChatsNew.from_dict(_chats_new) @@ -355,9 +323,7 @@ def getWatchChats( "receiver_name": chat.receiver.name, "data_text": chat.data.text, "data_sender_name": chat.data.sender_name, - "create": datetime.fromtimestamp(chat.create).strftime( - "%Y-%m-%d %H:%M:%S" - ), + "create": datetime.fromtimestamp(chat.create).strftime("%Y-%m-%d %H:%M:%S"), "delete_flag": chat.data.delete_flag, "emoticon_id": chat.data.emoticon_id, } @@ -408,11 +374,7 @@ def getWatchChatsRaw( result = ChatsNew.from_dict(result) - filtered_chats = [ - chat - for chat in result.list - if show_del_msg or chat.data.delete_flag == 0 - ] + filtered_chats = [chat for chat in result.list if show_del_msg or chat.data.delete_flag == 0] chats_new = ChatsNew(filtered_chats).to_dict() except Error as error: _LOGGER.debug(error) @@ -420,9 +382,7 @@ def getWatchChatsRaw( if not chats_new: self.delay(self.retryDelay) - return ( - ChatsNew.from_dict(chats_new, infer_missing=True) if asObject else chats_new - ) + return ChatsNew.from_dict(chats_new, infer_missing=True) if asObject else chats_new ##### Watch Location Info ##### def getWatchLastLocation(self, wuid: str, withAsk: bool = False) -> dict[str, Any]: @@ -532,9 +492,7 @@ def setDisableSilentTime(self, silent_id: str) -> bool: while not result and retry_counter < self.maxRetries + 2: retry_counter += 1 try: - disable_raw = self._gql_handler.setEnableSilentTime( - silent_id, NormalStatus.DISABLE.value - ) + disable_raw = self._gql_handler.setEnableSilentTime(silent_id, NormalStatus.DISABLE.value) result = disable_raw.get("setEnableSilentTime", False) except Error as error: _LOGGER.debug(error) @@ -600,10 +558,7 @@ def sendText(self, text: str, wuid: str) -> bool: def isAdmin(self, wuid: str) -> bool: user_id = self.getUserID() contacts = self.getWatchUserContacts(wuid) - return any( - contact["id"] == user_id and contact["guardianType"] == "FIRST" - for contact in contacts - ) + return any(contact["id"] == user_id and contact["guardianType"] == "FIRST" for contact in contacts) def shutdown(self, wuid: str) -> bool: if self.isAdmin(wuid): @@ -644,17 +599,13 @@ def getWatches(self, wuid: str) -> dict[str, Any]: self.delay(self.retryDelay) return watch - def getSWInfo( - self, wuid: str, watches: Optional[dict[str, Any]] = None - ) -> dict[str, Any]: + def getSWInfo(self, wuid: str, watches: Optional[dict[str, Any]] = None) -> dict[str, Any]: watches = {} if watches is None else watches wqr: dict[str, Any] = watches if watches else self.getWatches(wuid=wuid) qrCode: str = wqr.get("qrCode", "=") return self._gql_handler.getSWInfo(qrCode.split("=")[1]) - def getWatchState( - self, wuid: str, watches: Optional[dict[str, Any]] = None - ) -> dict[str, Any]: + def getWatchState(self, wuid: str, watches: Optional[dict[str, Any]] = None) -> dict[str, Any]: watches = {} if watches is None else watches wqr: dict[str, Any] = watches if watches else self.getWatches(wuid=wuid) qrCode: str = wqr.get("qrCode", "=") @@ -670,9 +621,7 @@ def getCountries(self) -> list[dict[str, str]]: countries: dict[str, Any] = self._gql_handler.countries() return countries.get("countries", {}) - def getWatchLocHistory( - self, wuid: str, date: int, tz: str, limit: int - ) -> dict[str, Any]: + def getWatchLocHistory(self, wuid: str, date: int, tz: str, limit: int) -> dict[str, Any]: return self._gql_handler.getWatchLocHistory(wuid, date, tz, limit) def watchesDynamic(self) -> dict[str, Any]: @@ -688,9 +637,7 @@ def avatars(self, id: str) -> dict[str, Any]: return self._gql_handler.avatars(id) def getWatchUserSteps(self, wuid: str, date: int) -> dict[str, Any]: - userSteps = self._gql_handler.getWatchUserSteps( - wuid=wuid, tz=self._timeZone, date=date - ) + userSteps = self._gql_handler.getWatchUserSteps(wuid=wuid, tz=self._timeZone, date=date) if not userSteps: return {} userSteps = userSteps.get("userSteps", {}) @@ -712,12 +659,8 @@ def addStep(self, step: int) -> bool: s: dict[str, bool] = self._gql_handler.addStep(step) return s.get("addStep", False) - def submitIncorrectLocationData( - self, wuid: str, lat: str, lng: str, timestamp: str - ) -> bool: - data: dict[str, bool] = self._gql_handler.submitIncorrectLocationData( - wuid, lat, lng, timestamp - ) + def submitIncorrectLocationData(self, wuid: str, lat: str, lng: str, timestamp: str) -> bool: + data: dict[str, bool] = self._gql_handler.submitIncorrectLocationData(wuid, lat, lng, timestamp) return data.get("submitIncorrectLocationData", False) def getAppVersion(self) -> dict[str, Any]: @@ -731,9 +674,7 @@ def checkEmailOrPhoneExist( countryCode: str = "", phoneNumber: str = "", ) -> bool: - data = self._gql_handler.checkEmailOrPhoneExist( - type, email, countryCode, phoneNumber - ) + data = self._gql_handler.checkEmailOrPhoneExist(type, email, countryCode, phoneNumber) return data.get("checkEmailOrPhoneExist", False) def modifyContact( diff --git a/src/pyxplora_api/pyxplora_api_async.py b/src/pyxplora_api/pyxplora_api_async.py index 39e6eac..b69be5a 100644 --- a/src/pyxplora_api/pyxplora_api_async.py +++ b/src/pyxplora_api/pyxplora_api_async.py @@ -67,9 +67,7 @@ def __init__( session, ) - async def _login( - self, force_login: bool = False, key=None, sec=None - ) -> tuple[dict[str, Any] | None, str | None]: + async def _login(self, force_login: bool = False, key=None, sec=None) -> tuple[dict[str, Any] | None, str | None]: if not self._isConnected() or self._hasTokenExpired() or force_login: retryCounter = 0 while not self._isConnected() and (retryCounter < self.maxRetries + 2): @@ -78,9 +76,7 @@ async def _login( # Try to login try: - self._issueToken, self._refresh_token = ( - await self._gql_handler.login_a(key, sec) - ) + self._issueToken, self._refresh_token = await self._gql_handler.login_a(key, sec) except LoginError as error: self.error_message = error.error_message await asyncio.sleep(self.retryDelay) @@ -94,9 +90,7 @@ async def _login( self.dtIssueToken = int(time()) return self._issueToken, self._refresh_token - async def init( - self, forceLogin: bool = False, signup: bool = True, key=None, sec=None - ) -> None: + async def init(self, forceLogin: bool = False, signup: bool = True, key=None, sec=None) -> None: # self.initHandler(signup) token, refresh_token = await self._login(forceLogin, key, sec) if not signup: @@ -120,11 +114,7 @@ async def init( if not self._childPhoneNumber: self.watchs = children else: - self.watchs = [ - watch - for watch in children - if watch["ward"]["phoneNumber"] in self._childPhoneNumber - ] + self.watchs = [watch for watch in children if watch["ward"]["phoneNumber"] in self._childPhoneNumber] self.user = user @staticmethod @@ -167,11 +157,7 @@ async def _setDevice(self, wuid: str) -> None: self.getSWInfo(wuid, watches=await self.getWatches(wuid)), self.getWatchUserSteps( wuid, - date=int( - datetime.now() - .replace(hour=0, minute=0, second=0, microsecond=0) - .timestamp() - ), + date=int(datetime.now().replace(hour=0, minute=0, second=0, microsecond=0).timestamp()), ), self.getWatchOnlineStatus(wuid), ] @@ -196,16 +182,12 @@ async def _setDevice(self, wuid: str) -> None: "watch_battery": battery, "watch_charging": isCharging, "locateType": locateType, - "lastTrackTime": watch_location.get( - "tm", datetime.now().strftime("%Y-%m-%d %H:%M:%S") - ), + "lastTrackTime": watch_location.get("tm", datetime.now().strftime("%Y-%m-%d %H:%M:%S")), "lat": watch_location.get("lat", None), "lng": watch_location.get("lng", None), "rad": watch_location.get("rad", -1), "step": watch_location.get("watch_last_location", {}).get("step", 0), - "distance": watch_location.get("watch_last_location", {}).get( - "distance", -1 - ), + "distance": watch_location.get("watch_last_location", {}).get("distance", -1), "isInSafeZone": isInSafeZone, "safeZoneLabel": safeZoneLabel, "getWatchSafeZones": watch_safe_zones, @@ -239,12 +221,8 @@ async def getWatchUserContacts(self, wuid: str) -> list[dict[str, Any]]: { "id": _id, "guardianType": contact["guardianType"], - "create": datetime.fromtimestamp( - contact["create"] - ).strftime("%Y-%m-%d %H:%M:%S"), - "update": datetime.fromtimestamp( - contact["update"] - ).strftime("%Y-%m-%d %H:%M:%S"), + "create": datetime.fromtimestamp(contact["create"]).strftime("%Y-%m-%d %H:%M:%S"), + "update": datetime.fromtimestamp(contact["update"]).strftime("%Y-%m-%d %H:%M:%S"), "name": contact["name"], "phoneNumber": f'+{contact["countryPhoneNumber"]}{contact["phoneNumber"]}', "xcoin": xcoin, @@ -284,9 +262,7 @@ async def getWatchAlarm(self, wuid: str) -> list[dict[str, Any]]: await asyncio.sleep(self.retryDelay) return alarms - async def loadWatchLocation( - self, wuid: str = "", with_ask: bool = True - ) -> dict[str, Any]: + async def loadWatchLocation(self, wuid: str = "", with_ask: bool = True) -> dict[str, Any]: retry_counter = 0 watch_location = {} while retry_counter < self.maxRetries + 1: @@ -301,11 +277,7 @@ async def loadWatchLocation( _watch_last_locate = location_raw.get("watchLastLocate", {}) if not _watch_last_locate: return watch_location - _tm = ( - 31532399 - if _watch_last_locate.get("tm") is None - else _watch_last_locate.get("tm") - ) + _tm = 31532399 if _watch_last_locate.get("tm") is None else _watch_last_locate.get("tm") _lat = _watch_last_locate.get("lat", "0.0") _lng = _watch_last_locate.get("lng", "0.0") _rad = _watch_last_locate.get("rad", -1) @@ -371,11 +343,7 @@ async def getWatchOnlineStatus(self, wuid: str) -> str: try: ask_raw = await self.askWatchLocate(wuid) track_raw = await self.getTrackWatchInterval(wuid) - status = ( - WatchOnlineStatus.ONLINE - if ask_raw or track_raw != -1 - else WatchOnlineStatus.OFFLINE - ) + status = WatchOnlineStatus.ONLINE if ask_raw or track_raw != -1 else WatchOnlineStatus.OFFLINE except Error as error: _LOGGER.debug(error) retries += 1 @@ -409,9 +377,7 @@ async def getWatchChats( while not chats and retry_counter < self.maxRetries + 2: retry_counter += 1 try: - _chats_new = await self.getWatchChatsRaw( - wuid, offset, limit, msgId, show_del_msg, asObject - ) + _chats_new = await self.getWatchChatsRaw(wuid, offset, limit, msgId, show_del_msg, asObject) if isinstance(_chats_new, dict): _chats_new = ChatsNew.from_dict(_chats_new) @@ -429,9 +395,7 @@ async def getWatchChats( "receiver_name": chat.receiver.name, "data_text": chat.data.text, "data_sender_name": chat.data.sender_name, - "create": datetime.fromtimestamp(chat.create).strftime( - "%Y-%m-%d %H:%M:%S" - ), + "create": datetime.fromtimestamp(chat.create).strftime("%Y-%m-%d %H:%M:%S"), "delete_flag": chat.data.delete_flag, "emoticon_id": chat.data.emoticon_id, } @@ -464,10 +428,8 @@ async def getWatchChatsRaw( while not chats_new and retry_counter < self.maxRetries + 2: retry_counter += 1 try: - result: dict[str, Any] | Chats | ChatsNew | str | None = ( - await self._gql_handler.chats_a( - wuid, offset, limit, msgId, asObject - ) + result: dict[str, Any] | Chats | ChatsNew | str | None = await self._gql_handler.chats_a( + wuid, offset, limit, msgId, asObject ) if not result: @@ -497,11 +459,7 @@ async def getWatchChatsRaw( d.data.emoticon_id = Emoji[f"M{d.data.emoticon_id}"].value await self.set_read_chat_msg(wuid, d.msgId, d.id) - filtered_chats = [ - chat - for chat in result.list - if show_del_msg or chat.data.delete_flag == 0 - ] + filtered_chats = [chat for chat in result.list if show_del_msg or chat.data.delete_flag == 0] chats_new = ChatsNew(filtered_chats).to_dict() except Error as error: _LOGGER.debug(error) @@ -509,9 +467,7 @@ async def getWatchChatsRaw( if not chats_new: await asyncio.sleep(self.retryDelay) - return ( - ChatsNew.from_dict(chats_new, infer_missing=True) if asObject else chats_new - ) + return ChatsNew.from_dict(chats_new, infer_missing=True) if asObject else chats_new ##### Watch Location Info ##### async def getWatchLastLocation(self, wuid: str) -> dict[str, Any]: @@ -570,9 +526,7 @@ async def getTrackWatchInterval(self, wuid: str) -> int: return (await self._gql_handler.trackWatch_a(wuid)).get("trackWatch", -1) async def askWatchLocate(self, wuid: str) -> bool: - return (await self._gql_handler.askWatchLocate_a(wuid)).get( - "askWatchLocate", False - ) + return (await self._gql_handler.askWatchLocate_a(wuid)).get("askWatchLocate", False) ##### Feature ##### async def getSilentTime(self, wuid: str) -> list[dict[str, Any]]: @@ -630,9 +584,7 @@ async def setDisableSilentTime(self, silent_id: str) -> bool: while not result and retry_counter < self.maxRetries + 2: retry_counter += 1 try: - disable_raw = await self._gql_handler.setEnableSilentTime_a( - silent_id, NormalStatus.DISABLE.value - ) + disable_raw = await self._gql_handler.setEnableSilentTime_a(silent_id, NormalStatus.DISABLE.value) result = disable_raw.get("setEnableSilentTime", False) except Error as error: _LOGGER.debug(error) @@ -662,9 +614,7 @@ async def setAlarmTime(self, alarm_id: str, status: NormalStatus) -> bool: while not result and (retryCounter < self.maxRetries + 2): retryCounter += 1 try: - raw = await self._gql_handler.setEnableAlarmTime_a( - alarm_id, status.value - ) + raw = await self._gql_handler.setEnableAlarmTime_a(alarm_id, status.value) modifyAlarm = raw.get("modifyAlarm", -1) if not modifyAlarm: return False @@ -700,10 +650,7 @@ async def sendText(self, text: str, wuid: str) -> bool: async def isAdmin(self, wuid: str) -> bool: user_id = self.getUserID() contacts = await self.getWatchUserContacts(wuid) - return any( - contact["id"] == user_id and contact["guardianType"] == "FIRST" - for contact in contacts - ) + return any(contact["id"] == user_id and contact["guardianType"] == "FIRST" for contact in contacts) async def shutdown(self, wuid: str) -> bool: if await self.isAdmin(wuid): @@ -744,17 +691,13 @@ async def getWatches(self, wuid: str) -> dict[str, Any]: await asyncio.sleep(self.retryDelay) return watch - async def getSWInfo( - self, wuid: str, watches: Optional[dict[str, Any]] = None - ) -> dict[str, Any]: + async def getSWInfo(self, wuid: str, watches: Optional[dict[str, Any]] = None) -> dict[str, Any]: watches = {} if watches is None else watches wqr: dict[str, Any] = watches if watches else await self.getWatches(wuid=wuid) qrCode: str = wqr.get("qrCode", "=") return await self._gql_handler.getSWInfo_a(qrCode.split("=")[1]) - async def getWatchState( - self, wuid: str, watches: Optional[dict[str, Any]] = None - ) -> dict[str, Any]: + async def getWatchState(self, wuid: str, watches: Optional[dict[str, Any]] = None) -> dict[str, Any]: watches = {} if watches is None else watches wqr: dict[str, Any] = watches if watches else await self.getWatches(wuid=wuid) qrCode: str = wqr.get("qrCode", "=") @@ -770,9 +713,7 @@ async def getCountries(self) -> list[dict[str, str]]: countries: dict[str, Any] = await self._gql_handler.countries_a() return countries.get("countries", {}) - async def getWatchLocHistory( - self, wuid: str, date: int, tz: str, limit: int - ) -> dict[str, Any]: + async def getWatchLocHistory(self, wuid: str, date: int, tz: str, limit: int) -> dict[str, Any]: return await self._gql_handler.getWatchLocHistory_a(wuid, date, tz, limit) async def watchesDynamic(self) -> dict[str, Any]: @@ -781,18 +722,14 @@ async def watchesDynamic(self) -> dict[str, Any]: async def watchGroups(self, _id: str = "") -> dict[str, Any]: return await self._gql_handler.watchGroups_a(_id) - async def familyInfo( - self, wuid: str, watchId: str, tz: str, date: int - ) -> dict[str, Any]: + async def familyInfo(self, wuid: str, watchId: str, tz: str, date: int) -> dict[str, Any]: return await self._gql_handler.familyInfo_a(wuid, watchId, tz, date) async def avatars(self, _id: str) -> dict[str, Any]: return await self._gql_handler.avatars_a(_id) async def getWatchUserSteps(self, wuid: str, date: int) -> dict[str, Any]: - userSteps = await self._gql_handler.getWatchUserSteps_a( - wuid=wuid, tz=self._timeZone, date=date - ) + userSteps = await self._gql_handler.getWatchUserSteps_a(wuid=wuid, tz=self._timeZone, date=date) if not userSteps: return {} userSteps = userSteps.get("userSteps", {}) @@ -814,12 +751,8 @@ async def addStep(self, step: int) -> bool: s: dict[str, bool] = await self._gql_handler.addStep_a(step) return s.get("addStep", False) - async def submitIncorrectLocationData( - self, wuid: str, lat: str, lng: str, timestamp: str - ) -> bool: - data: dict[str, bool] = await self._gql_handler.submitIncorrectLocationData_a( - wuid, lat, lng, timestamp - ) + async def submitIncorrectLocationData(self, wuid: str, lat: str, lng: str, timestamp: str) -> bool: + data: dict[str, bool] = await self._gql_handler.submitIncorrectLocationData_a(wuid, lat, lng, timestamp) return data.get("submitIncorrectLocationData", False) async def getAppVersion(self) -> dict[str, Any]: @@ -833,17 +766,11 @@ async def checkEmailOrPhoneExist( countryCode: str = "", phoneNumber: str = "", ) -> bool: - data = await self._gql_handler.checkEmailOrPhoneExist_a( - type, email, countryCode, phoneNumber - ) + data = await self._gql_handler.checkEmailOrPhoneExist_a(type, email, countryCode, phoneNumber) return data.get("checkEmailOrPhoneExist", False) - async def modifyContact( - self, contactId: str, isAdmin: bool, contactName: str = "", fileId: str = "" - ) -> dict[str, Any]: - data = await self._gql_handler.modifyContact_a( - contactId, isAdmin, contactName, fileId - ) + async def modifyContact(self, contactId: str, isAdmin: bool, contactName: str = "", fileId: str = "") -> dict[str, Any]: + data = await self._gql_handler.modifyContact_a(contactId, isAdmin, contactName, fileId) return data async def deleteMessageFromApp(self, wuid: str, msgId: str) -> bool: diff --git a/tests/test_exception_classes.py b/tests/test_exception_classes.py index ff51dee..6d91455 100644 --- a/tests/test_exception_classes.py +++ b/tests/test_exception_classes.py @@ -20,19 +20,12 @@ def test_exception_messages_are_human_readable() -> None: assert str(ChildNoError()) == "Child phonenumber & Watch ID not found!" assert str(ChildNoError(["Watch ID"])) == "Watch ID not found!" assert str(XTypeError("str", int)) == ( - "Transfer value has the wrong type! The following are permitted: str. " - "The specified type is: " - ) - assert ( - str(FunctionError("call")) - == "Xplora API call finally failed with response: call" + "Transfer value has the wrong type! The following are permitted: str. " "The specified type is: " ) + assert str(FunctionError("call")) == "Xplora API call finally failed with response: call" def test_login_and_contact_errors_accept_error_enums() -> None: assert str(LoginError(ErrorMSG.AUTH_FAIL)) == "Authentication failed." assert str(PhoneOrEmailFail()) == "Phone Number or Email address not exist" - assert ( - str(PhoneOrEmailFail(ErrorMSG.PHONE_MAIL_ERR)) - == "Phone Number or Email address not exist" - ) + assert str(PhoneOrEmailFail(ErrorMSG.PHONE_MAIL_ERR)) == "Phone Number or Email address not exist" diff --git a/tests/test_gql_handler_sync.py b/tests/test_gql_handler_sync.py index 7db1d69..9ceec9b 100644 --- a/tests/test_gql_handler_sync.py +++ b/tests/test_gql_handler_sync.py @@ -14,9 +14,7 @@ def __init__(self) -> None: self.next_response = {"data": {}} def runAuthorizedGqlQuery(self, query, variables=None, operation_name=None): - self.calls.append( - {"query": query, "variables": variables, "operation_name": operation_name} - ) + self.calls.append({"query": query, "variables": variables, "operation_name": operation_name}) return self.next_response @@ -33,9 +31,7 @@ def test_login_populates_token_fields(monkeypatch) -> None: monkeypatch.setattr( handler, "runGqlQuery", - lambda query, variables=None, operation_name=None: { - "data": {"signInWithEmailOrPhone": token} - }, + lambda query, variables=None, operation_name=None: {"data": {"signInWithEmailOrPhone": token}}, ) assert handler.login() == token @@ -70,11 +66,7 @@ def test_is_admin_returns_true_for_first_guardian_and_raises_otherwise( monkeypatch.setattr( handler, "getWatchUserContacts", - lambda wuid: { - "contacts": { - "contacts": [{"contactUser": {"id": "user-1"}, "guardianType": "FIRST"}] - } - }, + lambda wuid: {"contacts": {"contacts": [{"contactUser": {"id": "user-1"}, "guardianType": "FIRST"}]}}, ) monkeypatch.setattr( handler, @@ -86,13 +78,7 @@ def test_is_admin_returns_true_for_first_guardian_and_raises_otherwise( monkeypatch.setattr( handler, "getWatchUserContacts", - lambda wuid: { - "contacts": { - "contacts": [ - {"contactUser": {"id": "user-1"}, "guardianType": "SECOND"} - ] - } - }, + lambda wuid: {"contacts": {"contacts": [{"contactUser": {"id": "user-1"}, "guardianType": "SECOND"}]}}, ) with pytest.raises(NoAdminError): handler.isAdmin("wuid-1", "query", {"uid": "wuid-1"}, "Contacts") @@ -106,9 +92,7 @@ def test_common_query_wrappers_return_data_and_capture_variables() -> None: assert handler.calls[-1]["operation_name"] == "Countries" handler.next_response = {"data": {"setEnableSilentTime": True}} - assert handler.setEnableSilentTime("silent-1", NormalStatus.DISABLE.value) == { - "setEnableSilentTime": True - } + assert handler.setEnableSilentTime("silent-1", NormalStatus.DISABLE.value) == {"setEnableSilentTime": True} assert handler.calls[-1]["variables"] == { "silentId": "silent-1", "status": "DISABLE", @@ -119,9 +103,7 @@ def test_common_query_wrappers_return_data_and_capture_variables() -> None: assert handler.calls[-1]["variables"] == {"uid": "wuid-1", "text": "Hallo"} handler.next_response = {"data": {"checkEmailOrPhoneExist": False}} - assert handler.checkEmailOrPhoneExist( - UserContactType.EMAIL, "user@example.test" - ) == {"checkEmailOrPhoneExist": False} + assert handler.checkEmailOrPhoneExist(UserContactType.EMAIL, "user@example.test") == {"checkEmailOrPhoneExist": False} assert handler.calls[-1]["variables"] == { "type": "EMAIL", "email": "user@example.test", diff --git a/tests/test_graphql_client.py b/tests/test_graphql_client.py index b8a5917..ec29982 100644 --- a/tests/test_graphql_client.py +++ b/tests/test_graphql_client.py @@ -19,10 +19,10 @@ def json(self) -> dict: class AsyncPostContext: - def __init__(self, response: "AsyncResponse") -> None: + def __init__(self, response: AsyncResponse) -> None: self.response = response - async def __aenter__(self) -> "AsyncResponse": + async def __aenter__(self) -> AsyncResponse: return self.response async def __aexit__(self, exc_type, exc, tb) -> None: @@ -80,13 +80,9 @@ def fake_post(endpoint: str, *, json: dict, headers: dict, timeout: int, **optio return response monkeypatch.setattr("pyxplora_api.graphql_client.requests.post", fake_post) - client = GraphqlClient( - "https://example.test/graphql", headers={"Authorization": "base"}, verify=False - ) + client = GraphqlClient("https://example.test/graphql", headers={"Authorization": "base"}, verify=False) - assert client.execute( - "query", {"id": 1}, "Operation", headers={"X-Test": "yes"} - ) == {"data": {"ok": True}} + assert client.execute("query", {"id": 1}, "Operation", headers={"X-Test": "yes"}) == {"data": {"ok": True}} assert response.raised is True assert calls == [ { @@ -110,13 +106,9 @@ def fake_post(endpoint: str, *, json: dict, headers: dict, timeout: int, **optio def test_ha_execute_async_uses_supplied_session_and_default_user_agent() -> None: response = AsyncResponse({"data": {"ok": True}}) session = FakeSession(response) - client = GraphqlClient( - "https://example.test/graphql", headers={"Authorization": "base"} - ) + client = GraphqlClient("https://example.test/graphql", headers={"Authorization": "base"}) - result = asyncio.run( - client.ha_execute_async("query", {"id": 2}, "Operation", session=session) - ) + result = asyncio.run(client.ha_execute_async("query", {"id": 2}, "Operation", session=session)) assert result == {"data": {"ok": True}} assert response.raised is True diff --git a/tests/test_handler_gql.py b/tests/test_handler_gql.py index 90978ba..d450a8a 100644 --- a/tests/test_handler_gql.py +++ b/tests/test_handler_gql.py @@ -11,9 +11,7 @@ def make_handler() -> HandlerGQL: - return HandlerGQL( - "49", "15123456789", "secret", "de-DE", "Europe/Berlin", "user@example.test" - ) + return HandlerGQL("49", "15123456789", "secret", "de-DE", "Europe/Berlin", "user@example.test") def test_constructor_hashes_password_and_builds_login_variables() -> None: @@ -21,7 +19,7 @@ def test_constructor_hashes_password_and_builds_login_variables() -> None: assert handler.getApiKey() == API_KEY assert handler.getSecret() == API_SECRET - assert handler.passwordMD5 == hashlib.md5("secret".encode()).hexdigest() + assert handler.passwordMD5 == hashlib.md5(b"secret").hexdigest() assert handler.variables == { "countryPhoneNumber": "49", "phoneNumber": "15123456789", diff --git a/tests/test_models_and_status.py b/tests/test_models_and_status.py index 781d32f..7e5f4bf 100644 --- a/tests/test_models_and_status.py +++ b/tests/test_models_and_status.py @@ -70,9 +70,7 @@ def test_dataclass_defaults_and_serialization_round_trip() -> None: assert small.emoticon_id == Emoticon.UNKNOWN__.value assert small.delete_flag == 0 - assert ( - SmallChatList.from_dict(small_list.to_dict()).small_chat_list[0].msgId == "msg" - ) + assert SmallChatList.from_dict(small_list.to_dict()).small_chat_list[0].msgId == "msg" def test_selected_status_enum_values_are_stable() -> None: diff --git a/tests/test_pyxplora_core.py b/tests/test_pyxplora_core.py index 8581f75..552f9f7 100644 --- a/tests/test_pyxplora_core.py +++ b/tests/test_pyxplora_core.py @@ -7,9 +7,7 @@ def make_client(*, wuid=None) -> PyXplora: - client = PyXplora( - "49", "15123456789", "secret", "de-DE", "Europe/Berlin", wuid=wuid - ) + client = PyXplora("49", "15123456789", "secret", "de-DE", "Europe/Berlin", wuid=wuid) client.watchs = [ { "ward": { @@ -87,9 +85,7 @@ def test_watch_user_accessors_support_all_str_and_filtered_inputs() -> None: assert client.getWatchUserPhoneNumbers("wuid-1") == "111" assert client.getWatchUserNames() == ["Alice", "Bob"] assert client.getWatchUserNames("wuid-2") == "Bob" - assert ( - client.getWatchUserIcons("wuid-1") == "https://api.myxplora.com/file?id=file-1" - ) + assert client.getWatchUserIcons("wuid-1") == "https://api.myxplora.com/file?id=file-1" assert client.getWatchUserIcons(["wuid-1", "wuid-2"]) == [ "https://api.myxplora.com/file?id=file-1", "https://api.myxplora.com/file?id=file-2",