Sourcery Starbot ⭐ refactored Alc-Alc/starlite

docs/examples/contrib/jwt/using_jwt_auth.py

@@ -46,12 +46,7 @@ async def retrieve_user_handler(token: Token, connection: "ASGIConnection[Any, A
 @post("/login")
 async def login_handler(data: User) -> Response[User]:
     MOCK_DB[str(data.id)] = data
-    response = jwt_auth.login(identifier=str(data.id), response_body=data)
-
-    # you can do whatever you want to update the response instance here
-    # e.g. response.set_cookie(...)
-
-    return response
+    return jwt_auth.login(identifier=str(data.id), response_body=data)
 
 
 # We also have some other routes, for example:

docs/examples/contrib/jwt/using_jwt_cookie_auth.py

@@ -48,12 +48,7 @@ async def retrieve_user_handler(token: "Token", connection: "ASGIConnection[Any,
 @post("/login")
 async def login_handler(data: "User") -> "Response[User]":
     MOCK_DB[str(data.id)] = data
-    response = jwt_cookie_auth.login(identifier=str(data.id), response_body=data)
-
-    # you can do whatever you want to update the response instance here
-    # e.g. response.set_cookie(...)
-
-    return response
+    return jwt_cookie_auth.login(identifier=str(data.id), response_body=data)
 
 
 # We also have some other routes, for example:

docs/examples/contrib/jwt/using_oauth2_password_bearer.py

@@ -48,26 +48,14 @@ async def retrieve_user_handler(token: "Token", connection: "ASGIConnection[Any,
 @post("/login")
 async def login_handler(request: "Request[Any, Any, Any]", data: "User") -> "Response[OAuth2Login]":
     MOCK_DB[str(data.id)] = data
-    # if we do not define a response body, the login process will return a standard OAuth2 login response.  Note the `Response[OAuth2Login]` return type.
-    response = oauth2_auth.login(identifier=str(data.id))
-
-    # you can do whatever you want to update the response instance here
-    # e.g. response.set_cookie(...)
-
-    return response
+    return oauth2_auth.login(identifier=str(data.id))
 
 
 @post("/login_custom")
 async def login_custom_response_handler(data: "User") -> "Response[User]":
     MOCK_DB[str(data.id)] = data
 
-    # If you'd like to define a custom response body, use the `response_body` parameter.  Note the `Response[User]` return type.
-    response = oauth2_auth.login(identifier=str(data.id), response_body=data)
-
-    # you can do whatever you want to update the response instance here
-    # e.g. response.set_cookie(...)
-
-    return response
+    return oauth2_auth.login(identifier=str(data.id), response_body=data)
 
 
 # We also have some other routes, for example:

docs/examples/parameters/header_and_cookie_parameters.py

@@ -26,7 +26,7 @@ async def get_user(
     token: str = Parameter(header="X-API-KEY"),
     cookie: str = Parameter(cookie="my-cookie-param"),
 ) -> User:
-    if not (token == VALID_TOKEN and cookie == VALID_COOKIE_VALUE):
+    if token != VALID_TOKEN or cookie != VALID_COOKIE_VALUE:
         raise NotAuthorizedException
     return User.parse_obj(USER_DB[user_id])
 

docs/examples/plugins/sqlalchemy_1_plugin/sqlalchemy_async.py

@@ -55,10 +55,10 @@ async def get_company(company_id: int, async_session: AsyncSession) -> Company:
     If a company with that ID does not exist, return a 404 response
     """
     result = await async_session.scalars(select(Company).where(Company.id == company_id))
-    company: Optional[Company] = result.one_or_none()
-    if not company:
+    if company := result.one_or_none():
+        return company
+    else:
         raise HTTPException(detail=f"Company with ID {company_id} not found", status_code=HTTP_404_NOT_FOUND)
-    return company
 
 
 app = Starlite(

docs/examples/plugins/sqlalchemy_1_plugin/sqlalchemy_relationships.py

@@ -46,10 +46,12 @@ def get_user(user_id: int, db_session: Session) -> User:
 
     If a user with that ID does not exist, return a 404 response
     """
-    user: Optional[User] = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
-    if not user:
+    if user := db_session.scalars(
+        select(User).where(User.id == user_id)
+    ).one_or_none():
+        return user
+    else:
         raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
-    return user
 
 
 app = Starlite(

docs/examples/plugins/sqlalchemy_1_plugin/sqlalchemy_relationships_to_many.py

@@ -64,10 +64,12 @@ def get_user(user_id: int, db_session: Session) -> UserModel:
 
     If a user with that ID does not exist, return a 404 response
     """
-    user: Optional[User] = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
-    if not user:
+    if user := db_session.scalars(
+        select(User).where(User.id == user_id)
+    ).one_or_none():
+        return UserModel.from_orm(user)
+    else:
         raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
-    return UserModel.from_orm(user)
 
 
 app = Starlite(

docs/examples/plugins/sqlalchemy_1_plugin/sqlalchemy_sync.py

@@ -50,10 +50,12 @@ def get_company(company_id: str, db_session: Session) -> Company:
 
     If a company with that ID does not exist, return a 404 response
     """
-    company: Optional[Company] = db_session.scalars(select(Company).where(Company.id == company_id)).one_or_none()
-    if not company:
+    if company := db_session.scalars(
+        select(Company).where(Company.id == company_id)
+    ).one_or_none():
+        return company
+    else:
         raise HTTPException(detail=f"Company with ID {company_id} not found", status_code=HTTP_404_NOT_FOUND)
-    return company
 
 
 app = Starlite(

docs/examples/security/using_session_auth.py

@@ -56,13 +56,7 @@ class UserLoginPayload(BaseModel):
 async def retrieve_user_handler(
     session: Dict[str, Any], connection: "ASGIConnection[Any, Any, Any, Any]"
 ) -> Optional[User]:
-    # we retrieve the user instance based on session data
-
-    user_id = session.get("user_id")
-    if user_id:
-        return MOCK_DB.get(user_id)
-
-    return None
+    return MOCK_DB.get(user_id) if (user_id := session.get("user_id")) else None
 
 
 @post("/login")

starlite/_asgi/routing_trie/mapping.py

@@ -85,10 +85,11 @@ def add_route_to_trie(
     """
     current_node = root_node
 
-    is_mount = hasattr(route, "route_handler") and getattr(route.route_handler, "is_mount", False)  # pyright: ignore
     has_path_parameters = bool(route.path_parameters)
 
-    if is_mount:  # pyright: ignore
+    if is_mount := hasattr(route, "route_handler") and getattr(
+        route.route_handler, "is_mount", False
+    ):
         current_node = add_mount_route(
             current_node=current_node,
             mount_routes=mount_routes,

starlite/_asgi/routing_trie/traversal.py

@@ -143,7 +143,9 @@ def parse_path_to_route(
             remaining_path = path[match.end() :]
             # since we allow regular handlers under static paths, we must validate that the request does not match
             # any such handler.
-            if not mount_node.children or not any(sub_route in path for sub_route in mount_node.children):  # type: ignore
+            if not mount_node.children or all(
+                sub_route not in path for sub_route in mount_node.children
+            ):  # type: ignore
                 asgi_app, handler = parse_node_handlers(node=mount_node, method=method)
                 remaining_path = remaining_path or "/"
                 if not mount_node.is_static:

starlite/_asgi/routing_trie/validate.py

@@ -31,8 +31,7 @@ def validate_node(node: RouteTrieNode) -> None:
         node.is_mount
         and node.children
         and any(
-            v
-            for v in chain.from_iterable(
+            chain.from_iterable(
                 list(child.path_parameters.values())
                 if isinstance(child.path_parameters, dict)
                 else child.path_parameters

starlite/_kwargs/kwargs_model.py

@@ -453,8 +453,11 @@ def _validate_raw_kwargs(
                     f"Make sure to use distinct keys for your dependencies, path parameters and aliased parameters."
                 )
 
-        used_reserved_kwargs = {*parameter_names, *path_parameters, *dependency_keys}.intersection(RESERVED_KWARGS)
-        if used_reserved_kwargs:
+        if used_reserved_kwargs := {
+            *parameter_names,
+            *path_parameters,
+            *dependency_keys,
+        }.intersection(RESERVED_KWARGS):
             raise ImproperlyConfiguredException(
                 f"Reserved kwargs ({', '.join(RESERVED_KWARGS)}) cannot be used for dependencies and parameter arguments. "
                 f"The following kwargs have been used: {', '.join(used_reserved_kwargs)}"

starlite/_kwargs/parameter_definition.py

@@ -60,7 +60,9 @@ def create_parameter_definition(
         field_alias=field_alias,
         default_value=default_value,
         is_required=signature_field.is_required
-        and (default_value is None and not (signature_field.is_optional or signature_field.is_any)),
+        and default_value is None
+        and not signature_field.is_optional
+        and not signature_field.is_any,
         is_sequence=signature_field.is_non_string_sequence,
     )
 

starlite/_multipart.py

@@ -123,7 +123,7 @@ def parse_multipart_form(body: bytes, boundary: bytes, multipart_form_part_limit
             line_index = line_end_index + 2
             colon_index = form_line.index(":")
             current_idx = colon_index + 2
-            form_header_field = form_line[0:colon_index].lower()
+            form_header_field = form_line[:colon_index].lower()
             form_header_value, form_parameters = parse_content_header(form_line[current_idx:])
 
             if form_header_field == "content-disposition":

starlite/_openapi/schema_generation/constrained_fields.py

@@ -12,7 +12,6 @@
 if TYPE_CHECKING:
     from starlite.plugins import OpenAPISchemaPluginProtocol
 
-if TYPE_CHECKING:
     from starlite._signature.models import SignatureField
 
     try:
@@ -158,10 +157,7 @@ def create_collection_constrained_field_schema(
             create_schema(field=sub_field, generate_examples=False, plugins=plugins, schemas=schemas)
             for sub_field in children
         ]
-        if len(items) > 1:
-            schema.items = Schema(one_of=items)
-        else:
-            schema.items = items[0]
+        schema.items = Schema(one_of=items) if len(items) > 1 else items[0]
     else:
         from starlite._signature.models import SignatureField
 

starlite/_openapi/typescript_converter/schema_parsing.py

@@ -50,12 +50,14 @@ def normalize_typescript_namespace(value: str, allow_quoted: bool) -> str:
     Returns:
         A normalized value
     """
-    if not allow_quoted and not (value[0].isalpha() or value[0] in {"_", "$"}):
+    if (
+        not allow_quoted
+        and not value[0].isalpha()
+        and value[0] not in {"_", "$"}
+    ):
         raise ValueError(f"invalid typescript namespace {value}")
     if allow_quoted:
-        if allowed_key_re.fullmatch(value):
-            return value
-        return f'"{value}"'
+        return value if allowed_key_re.fullmatch(value) else f'"{value}"'
     return invalid_namespace_re.sub("", value)
 
 

starlite/_openapi/typescript_converter/types.py

@@ -24,15 +24,12 @@
 
 def _as_string(value: Any) -> str:
     if isinstance(value, str):
-        return '"' + value + '"'
+        return f'"{value}"'
 
     if isinstance(value, bool):
         return "true" if value else "false"
 
-    if value is None:
-        return "null"
-
-    return str(value)
+    return "null" if value is None else str(value)
 
 
 class TypeScriptElement(ABC):

starlite/_parsers.py

@@ -48,10 +48,14 @@ def parse_cookie_string(cookie_string: str) -> dict[str, str]:
     Returns:
         A string keyed dictionary of values
     """
-    output: dict[str, str] = {}
     cookies = [cookie.split("=", 1) if "=" in cookie else ("", cookie) for cookie in cookie_string.split(";")]
-    for k, v in filter(lambda x: x[0] or x[1], ((k.strip(), v.strip()) for k, v in cookies)):
-        output[k] = unquote(unquote_cookie(v))
+    output: dict[str, str] = {
+        k: unquote(unquote_cookie(v))
+        for k, v in filter(
+            lambda x: x[0] or x[1],
+            ((k.strip(), v.strip()) for k, v in cookies),
+        )
+    }
     return output
 
 

starlite/_signature/models.py

@@ -139,7 +139,11 @@ def is_required(self) -> bool:
         if isinstance(self.kwarg_model, ParameterKwarg) and self.kwarg_model.required is not None:
             return self.kwarg_model.required
 
-        return not (self.is_optional or self.is_any) and (self.is_empty or self.default_value is None)
+        return (
+            not self.is_optional
+            and not self.is_any
+            and (self.is_empty or self.default_value is None)
+        )
 
     @property
     def is_literal(self) -> bool:

starlite/_signature/parsing.py

@@ -166,9 +166,7 @@ def parse_fn_signature(
             )
 
         if isinstance(parameter.default, DependencyKwarg) and parameter.name not in dependency_name_set:
-            if not parameter.optional and (
-                isinstance(parameter.default, DependencyKwarg) and parameter.default.default is Empty
-            ):
+            if not parameter.optional and parameter.default.default is Empty:
                 raise ImproperlyConfiguredException(
                     f"Explicit dependency '{parameter.name}' for '{fn_name}' has no default value, "
                     f"or provided dependency."

starlite/cli/_utils.py

@@ -291,12 +291,10 @@ def _autodiscover_app(cwd: Path) -> LoadedApp:
 
 def _format_is_enabled(value: Any) -> str:
     """Return a coloured string `"Enabled" if ``value`` is truthy, else "Disabled"."""
-    if value:
-        return "[green]Enabled[/]"
-    return "[red]Disabled[/]"
+    return "[green]Enabled[/]" if value else "[red]Disabled[/]"
 
 
-def show_app_info(app: Starlite) -> None:  # pragma: no cover
+def show_app_info(app: Starlite) -> None:    # pragma: no cover
     """Display basic information about the application and its configuration."""
 
     table = Table(show_header=False)
@@ -324,12 +322,10 @@ def show_app_info(app: Starlite) -> None:  # pragma: no cover
 
     if app.static_files_config:
         static_files_configs = app.static_files_config
-        static_files_info = []
-        for static_files in static_files_configs:
-            static_files_info.append(
-                f"path=[yellow]{static_files.path}[/] dirs=[yellow]{', '.join(map(str, static_files.directories))}[/] "
-                f"html_mode={_format_is_enabled(static_files.html_mode)}",
-            )
+        static_files_info = [
+            f"path=[yellow]{static_files.path}[/] dirs=[yellow]{', '.join(map(str, static_files.directories))}[/] html_mode={_format_is_enabled(static_files.html_mode)}"
+            for static_files in static_files_configs
+        ]
         table.add_row("Static files", "\n".join(static_files_info))
 
     if app.serialization_plugins:

starlite/connection/base.py

@@ -184,9 +184,7 @@ def cookies(self) -> dict[str, str]:
         """
         if self._cookies is Empty:
             cookies: dict[str, str] = {}
-            cookie_header = self.headers.get("cookie")
-
-            if cookie_header:
+            if cookie_header := self.headers.get("cookie"):
                 cookies = parse_cookie_string(cookie_header)
 
             self._cookies = self.scope["_cookies"] = cookies  # type: ignore[typeddict-unknown-key]