'Refactored by Sourcery'

Author: Sourcery AI

examples/decorators/provide_session.py

@@ -46,11 +46,10 @@ def decorator(func: t.Callable[..., RT]) -> t.Callable[..., RT]:
         def wrapper(*args, **kwargs) -> RT:
             if "session" in kwargs or session_args_idx < len(args):
                 return func(*args, **kwargs)
-            else:
-                bind = Bind.get_instance(bind_name)
+            bind = Bind.get_instance(bind_name)
 
-                with create_session(bind) as session:
-                    return func(*args, session=session, **kwargs)
+            with create_session(bind) as session:
+                return func(*args, session=session, **kwargs)
 
         return wrapper
 

examples/usrsrv/component/repository.py

@@ -53,10 +53,12 @@ def __init__(self, session: Session):
         self._query = select(EntityMapper)
 
     def get(self, entity_id: EntityID) -> Entity:
-        dto = self._session.scalars(self._query.filter_by(uuid=entity_id)).one_or_none()
-        if not dto:
+        if dto := self._session.scalars(
+            self._query.filter_by(uuid=entity_id)
+        ).one_or_none():
+            return Entity(dto)
+        else:
             raise NotFound(entity_id)
-        return Entity(dto)
 
     def save(self, entity: Entity) -> None:
         self._session.add(entity.dto)

src/quart_sqlalchemy/config.py

@@ -146,8 +146,9 @@ class EngineConfig(ConfigBase):
     @root_validator
     def scrub_execution_options(cls, values):
         if "execution_options" in values:
-            execute_options = values["execution_options"].dict(exclude_defaults=True)
-            if execute_options:
+            if execute_options := values["execution_options"].dict(
+                exclude_defaults=True
+            ):
                 values["execution_options"] = execute_options
         return values
 

src/quart_sqlalchemy/framework/extension.py

@@ -14,7 +14,7 @@ def __init__(
         config: t.Optional[SQLAlchemyConfig] = None,
         app: t.Optional[Quart] = None,
     ):
-        initialize = False if config is None else True
+        initialize = config is not None
         super().__init__(config, initialize=initialize)
 
         if app is not None:

src/quart_sqlalchemy/model/mixins.py

@@ -301,7 +301,7 @@ def accumulate_mappings(class_, attribute) -> t.Dict[str, t.Any]:
         if base_class is class_:
             continue
         args = getattr(base_class, attribute, {})
-        accumulated.update(args)
+        accumulated |= args
 
     return accumulated
 
@@ -316,7 +316,7 @@ def accumulate_tuples_with_mapping(class_, attribute) -> t.Sequence[t.Any]:
         args = getattr(base_class, attribute, ())
         for arg in args:
             if isinstance(arg, t.Mapping):
-                accumulated_map.update(arg)
+                accumulated_map |= arg
             else:
                 accumulated_args.append(arg)
 

src/quart_sqlalchemy/session.py

@@ -45,17 +45,17 @@ def provide_global_contextual_session(func):
     @wraps(func)
     def wrapper(self, *args, **kwargs):
         session_in_args = any(
-            [isinstance(arg, (sa.orm.Session, sa.ext.asyncio.AsyncSession)) for arg in args]
+            isinstance(arg, (sa.orm.Session, sa.ext.asyncio.AsyncSession))
+            for arg in args
         )
         session_in_kwargs = "session" in kwargs
         session_provided = session_in_args or session_in_kwargs
 
         if session_provided:
             return func(self, *args, **kwargs)
-        else:
-            session = session_proxy()
+        session = session_proxy()
 
-            return func(self, session, *args, **kwargs)
+        return func(self, session, *args, **kwargs)
 
     return wrapper
 

src/quart_sqlalchemy/sim/auth.py

@@ -228,7 +228,7 @@ def auth_endpoint_security(self):
             results = self.authenticator.enforce(security_schemes, session)
             authorized_credentials = {}
             for result in results:
-                authorized_credentials.update(result)
+                authorized_credentials |= result
             g.authorized_credentials = authorized_credentials
 
 

src/quart_sqlalchemy/sim/config.py

@@ -16,21 +16,22 @@
 sa = sqlalchemy
 
 
+
+
 class AppSettings(BaseSettings):
+
     class Config:
         env_file = ".env", ".secrets.env"
 
     LOAD_BLUEPRINTS: t.List[str] = Field(
-        default_factory=lambda: list(("quart_sqlalchemy.sim.views.api",))
+        default_factory=lambda: ["quart_sqlalchemy.sim.views.api"]
     )
     LOAD_EXTENSIONS: t.List[str] = Field(
-        default_factory=lambda: list(
-            (
-                "quart_sqlalchemy.sim.db.db",
-                "quart_sqlalchemy.sim.app.schema",
-                "quart_sqlalchemy.sim.auth.auth",
-            )
-        )
+        default_factory=lambda: [
+            "quart_sqlalchemy.sim.db.db",
+            "quart_sqlalchemy.sim.app.schema",
+            "quart_sqlalchemy.sim.auth.auth",
+        ]
     )
     SECURITY_SCHEMES: t.Dict[str, SecuritySchemeBase] = Field(
         default_factory=lambda: {
@@ -52,4 +53,5 @@ class Config:
     WEB3_HTTPS_PROVIDER_URI: str = Field(env="WEB3_HTTPS_PROVIDER_URI")
 
 
+
 settings = AppSettings()

src/quart_sqlalchemy/sim/db.py

@@ -51,19 +51,13 @@ def process_bind_param(self, value, dialect):
         """Data going into to the database will be transformed by this method.
         See ``ObjectID`` for the design and rational for this.
         """
-        if value is None:
-            return None
-
-        return ObjectID(value).decode()
+        return None if value is None else ObjectID(value).decode()
 
     def process_result_value(self, value, dialect):
         """Data going out from the database will be explicitly casted to the
         ``ObjectID``.
         """
-        if value is None:
-            return None
-
-        return ObjectID(value)
+        return None if value is None else ObjectID(value)
 
 
 class MyBase(BaseMixins, sa.orm.DeclarativeBase):

src/quart_sqlalchemy/sim/handle.py

@@ -96,18 +96,13 @@ def update_app_name_by_id(self, magic_client_id, app_name):
             self.session_factory(), magic_client_id, app_name=app_name
         )
 
-        if not client:
-            return None
-
-        return client.app_name
+        return client.app_name if client else None
 
     def update_by_id(self, magic_client_id, **kwargs):
-        client = self.logic.MagicClient.update_by_id(
+        return self.logic.MagicClient.update_by_id(
             self.session_factory(), magic_client_id, **kwargs
         )
 
-        return client
-
     def set_inactive_by_id(self, magic_client_id):
         """
         Args:
@@ -309,16 +304,14 @@ def search_by_client_id_and_substring(
         if not isinstance(substring, str) or len(substring) < 3:
             raise InvalidSubstringError()
 
-        auth_users = self.logic.AuthUser.get_by_client_id_with_substring_search(
+        return self.logic.AuthUser.get_by_client_id_with_substring_search(
             self.session_factory(),
             client_id,
             substring,
             offset=offset,
             limit=limit,
         )
 
-        return auth_users
-
     def is_magic_connect_enabled(self, auth_user_id=None, auth_user=None):
         if auth_user is None and auth_user_id is None:
             raise Exception("At least one argument needed: auth_user_id or auth_user.")
@@ -397,11 +390,10 @@ def sync_auth_wallet(
     ):
         session = self.session_factory()
         with session.begin_nested():
-            existing_wallet = self.logic.AuthWallet.get_by_auth_user_id(
+            if existing_wallet := self.logic.AuthWallet.get_by_auth_user_id(
                 session,
                 auth_user_id,
-            )
-            if existing_wallet:
+            ):
                 raise RuntimeError("WalletExistsForNetworkAndWalletType")
 
             return self.logic.AuthWallet.add(