chore: more linting rules

Author: o-santi

src/auth/pyproject.toml

@@ -59,7 +59,7 @@ select = [
   # pyupgrade
   "UP",
   # flake8-bugbear
-  # "B",
+  "B",
   # flake8-simplify
   # "SIM",
   # isort

src/auth/src/supabase_auth/_async/gotrue_admin_api.py

@@ -41,15 +41,16 @@ def __init__(
         self,
         *,
         url: str = "",
-        headers: Dict[str, str] = {},
+        headers: Optional[Dict[str, str]] = None,
         http_client: Optional[AsyncClient] = None,
         verify: bool = True,
         proxy: Optional[str] = None,
     ) -> None:
+        http_headers = headers or {}
         AsyncGoTrueBaseAPI.__init__(
             self,
             url=url,
-            headers=headers,
+            headers=http_headers,
             http_client=http_client,
             verify=verify,
             proxy=proxy,
@@ -81,16 +82,17 @@ async def sign_out(self, jwt: str, scope: SignOutScope = "global") -> None:
     async def invite_user_by_email(
         self,
         email: str,
-        options: InviteUserByEmailOptions = {},
+        options: Optional[InviteUserByEmailOptions] = None,
     ) -> UserResponse:
         """
         Sends an invite link to an email address.
         """
+        email_options = options or {}
         response = await self._request(
             "POST",
             "invite",
-            body={"email": email, "data": options.get("data")},
-            redirect_to=options.get("redirect_to"),
+            body={"email": email, "data": email_options.get("data")},
+            redirect_to=email_options.get("redirect_to"),
         )
         return parse_user_response(response)
 

src/auth/src/supabase_auth/_async/gotrue_base_api.py

@@ -74,4 +74,4 @@ async def _request(
             response.raise_for_status()
             return response
         except (HTTPStatusError, RuntimeError) as e:
-            raise handle_exception(e)
+            raise handle_exception(e)  # noqa

src/auth/src/supabase_auth/_async/gotrue_client.py

@@ -673,19 +673,20 @@ async def get_user(self, jwt: Optional[str] = None) -> Optional[UserResponse]:
         return parse_user_response(await self._request("GET", "user", jwt=jwt))
 
     async def update_user(
-        self, attributes: UserAttributes, options: UpdateUserOptions = {}
+        self, attributes: UserAttributes, options: Optional[UpdateUserOptions]
     ) -> UserResponse:
         """
         Updates user data, if there is a logged in user.
         """
         session = await self.get_session()
         if not session:
             raise AuthSessionMissingError()
+        update_options = options or {}
         response = await self._request(
             "PUT",
             "user",
             body=attributes,
-            redirect_to=options.get("email_redirect_to"),
+            redirect_to=update_options.get("email_redirect_to"),
             jwt=session.access_token,
         )
         user_response = parse_user_response(response)
@@ -761,7 +762,7 @@ async def refresh_session(
         session = await self._call_refresh_token(refresh_token)
         return AuthResponse(session=session, user=session.user)
 
-    async def sign_out(self, options: SignOutOptions = {"scope": "global"}) -> None:
+    async def sign_out(self, options: Optional[SignOutOptions] = None) -> None:
         """
         `sign_out` will remove the logged in user from the
         current session and log them out - removing all items from storage and then trigger a `"SIGNED_OUT"` event.
@@ -771,13 +772,14 @@ async def sign_out(self, options: SignOutOptions = {"scope": "global"}) -> None:
         There is no way to revoke a user's access token jwt until it expires.
         It is recommended to set a shorter expiry on the jwt for this reason.
         """
+        signout_options = options or {"scope": "global"}
         with suppress(AuthApiError):
             session = await self.get_session()
             access_token = session.access_token if session else None
             if access_token:
-                await self.admin.sign_out(access_token, options["scope"])
+                await self.admin.sign_out(access_token, signout_options["scope"])
 
-        if options["scope"] != "others":
+        if signout_options["scope"] != "others":
             await self._remove_session()
             self._notify_all_subscribers("SIGNED_OUT", None)
 
@@ -801,31 +803,35 @@ def _unsubscribe() -> None:
         self._state_change_emitters[unique_id] = subscription
         return subscription
 
-    async def reset_password_for_email(self, email: str, options: Options = {}) -> None:
+    async def reset_password_for_email(
+        self, email: str, options: Optional[Options] = None
+    ) -> None:
         """
         Sends a password reset request to an email address.
         """
+        reset_options = options or {}
         await self._request(
             "POST",
             "recover",
             body={
                 "email": email,
                 "gotrue_meta_security": {
-                    "captcha_token": options.get("captcha_token"),
+                    "captcha_token": reset_options.get("captcha_token"),
                 },
             },
-            redirect_to=options.get("redirect_to"),
+            redirect_to=reset_options.get("redirect_to"),
         )
 
     async def reset_password_email(
         self,
         email: str,
-        options: Options = {},
+        options: Optional[Options] = None,
     ) -> None:
         """
         Sends a password reset request to an email address.
         """
-        await self.reset_password_for_email(email, options)
+
+        await self.reset_password_for_email(email, options or {})
 
     # MFA methods
 

src/auth/src/supabase_auth/_sync/gotrue_admin_api.py

@@ -41,15 +41,16 @@ def __init__(
         self,
         *,
         url: str = "",
-        headers: Dict[str, str] = {},
+        headers: Optional[Dict[str, str]] = None,
         http_client: Optional[SyncClient] = None,
         verify: bool = True,
         proxy: Optional[str] = None,
     ) -> None:
+        http_headers = headers or {}
         SyncGoTrueBaseAPI.__init__(
             self,
             url=url,
-            headers=headers,
+            headers=http_headers,
             http_client=http_client,
             verify=verify,
             proxy=proxy,
@@ -81,16 +82,17 @@ def sign_out(self, jwt: str, scope: SignOutScope = "global") -> None:
     def invite_user_by_email(
         self,
         email: str,
-        options: InviteUserByEmailOptions = {},
+        options: Optional[InviteUserByEmailOptions] = None,
     ) -> UserResponse:
         """
         Sends an invite link to an email address.
         """
+        email_options = options or {}
         response = self._request(
             "POST",
             "invite",
-            body={"email": email, "data": options.get("data")},
-            redirect_to=options.get("redirect_to"),
+            body={"email": email, "data": email_options.get("data")},
+            redirect_to=email_options.get("redirect_to"),
         )
         return parse_user_response(response)
 

src/auth/src/supabase_auth/_sync/gotrue_base_api.py

@@ -74,4 +74,4 @@ def _request(
             response.raise_for_status()
             return response
         except (HTTPStatusError, RuntimeError) as e:
-            raise handle_exception(e)
+            raise handle_exception(e)  # noqa

src/auth/src/supabase_auth/_sync/gotrue_client.py

@@ -669,19 +669,20 @@ def get_user(self, jwt: Optional[str] = None) -> Optional[UserResponse]:
         return parse_user_response(self._request("GET", "user", jwt=jwt))
 
     def update_user(
-        self, attributes: UserAttributes, options: UpdateUserOptions = {}
+        self, attributes: UserAttributes, options: Optional[UpdateUserOptions]
     ) -> UserResponse:
         """
         Updates user data, if there is a logged in user.
         """
         session = self.get_session()
         if not session:
             raise AuthSessionMissingError()
+        update_options = options or {}
         response = self._request(
             "PUT",
             "user",
             body=attributes,
-            redirect_to=options.get("email_redirect_to"),
+            redirect_to=update_options.get("email_redirect_to"),
             jwt=session.access_token,
         )
         user_response = parse_user_response(response)
@@ -755,7 +756,7 @@ def refresh_session(self, refresh_token: Optional[str] = None) -> AuthResponse:
         session = self._call_refresh_token(refresh_token)
         return AuthResponse(session=session, user=session.user)
 
-    def sign_out(self, options: SignOutOptions = {"scope": "global"}) -> None:
+    def sign_out(self, options: Optional[SignOutOptions] = None) -> None:
         """
         `sign_out` will remove the logged in user from the
         current session and log them out - removing all items from storage and then trigger a `"SIGNED_OUT"` event.
@@ -765,13 +766,14 @@ def sign_out(self, options: SignOutOptions = {"scope": "global"}) -> None:
         There is no way to revoke a user's access token jwt until it expires.
         It is recommended to set a shorter expiry on the jwt for this reason.
         """
+        signout_options = options or {"scope": "global"}
         with suppress(AuthApiError):
             session = self.get_session()
             access_token = session.access_token if session else None
             if access_token:
-                self.admin.sign_out(access_token, options["scope"])
+                self.admin.sign_out(access_token, signout_options["scope"])
 
-        if options["scope"] != "others":
+        if signout_options["scope"] != "others":
             self._remove_session()
             self._notify_all_subscribers("SIGNED_OUT", None)
 
@@ -795,31 +797,35 @@ def _unsubscribe() -> None:
         self._state_change_emitters[unique_id] = subscription
         return subscription
 
-    def reset_password_for_email(self, email: str, options: Options = {}) -> None:
+    def reset_password_for_email(
+        self, email: str, options: Optional[Options] = None
+    ) -> None:
         """
         Sends a password reset request to an email address.
         """
+        reset_options = options or {}
         self._request(
             "POST",
             "recover",
             body={
                 "email": email,
                 "gotrue_meta_security": {
-                    "captcha_token": options.get("captcha_token"),
+                    "captcha_token": reset_options.get("captcha_token"),
                 },
             },
-            redirect_to=options.get("redirect_to"),
+            redirect_to=reset_options.get("redirect_to"),
         )
 
     def reset_password_email(
         self,
         email: str,
-        options: Options = {},
+        options: Optional[Options] = None,
     ) -> None:
         """
         Sends a password reset request to an email address.
         """
-        self.reset_password_for_email(email, options)
+
+        self.reset_password_for_email(email, options or {})
 
     # MFA methods
 

src/auth/src/supabase_auth/helpers.py

@@ -226,8 +226,8 @@ def decode_jwt(token: str) -> DecodedJWT:
         header = base64url_to_bytes(parts[0])
         payload = base64url_to_bytes(parts[1])
         signature = base64url_to_bytes(parts[2])
-    except binascii.Error:
-        raise AuthInvalidJwtError("Invalid JWT structure")
+    except binascii.Error as e:
+        raise AuthInvalidJwtError("Invalid JWT structure") from e
 
     return DecodedJWT(
         header=JWTHeaderParser.validate_json(header),

src/auth/tests/_async/clients.py

@@ -35,14 +35,15 @@ class Credentials:
 
 
 def mock_user_credentials(
-    options: OptionalCredentials = {},
+    options: Optional[OptionalCredentials] = None,
 ) -> Credentials:
     fake = Faker()
+    user_options = options or {}
     rand_numbers = str(int(time()))
     return Credentials(
-        email=options.get("email") or fake.email(),
-        phone=options.get("phone") or f"1{rand_numbers[-11:]}",
-        password=options.get("password") or fake.password(),
+        email=user_options.get("email") or fake.email(),
+        phone=user_options.get("phone") or f"1{rand_numbers[-11:]}",
+        password=user_options.get("password") or fake.password(),
     )
 
 

src/auth/tests/_async/test_gotrue.py

@@ -464,14 +464,16 @@ async def test_sign_in_with_password() -> None:
                 "password": "wrong_password",
             }
         )
-        assert False, "Expected AuthApiError for wrong password"
+        raise AssertionError("Expected AuthApiError for wrong password")
     except AuthApiError:
         pass
 
     # Test error case: missing credentials
     try:
         await test_client.sign_in_with_password({})  # type: ignore
-        assert False, "Expected AuthInvalidCredentialsError for missing credentials"
+        raise AssertionError(
+            "Expected AuthInvalidCredentialsError for missing credentials"
+        )
     except AuthInvalidCredentialsError:
         pass
 
@@ -569,7 +571,7 @@ async def test_sign_in_with_otp() -> None:
 
     try:
         await client.sign_in_with_otp({})  # type: ignore
-        assert False, "Expected AuthInvalidCredentialsError"
+        raise AssertionError("Expected AuthInvalidCredentialsError")
     except AuthInvalidCredentialsError:
         pass