ITISFoundation · matusdrobuliak66 · Apr 24, 2023 · Apr 22, 2023 · Apr 22, 2023 · Apr 22, 2023
@@ -26,7 +26,7 @@
 from .login.plugin import setup_login
 from .long_running_tasks import setup_long_running_tasks
 from .meta_modeling.plugin import setup_meta_modeling
-from .products import setup_products
+from .products.plugin import setup_products
 from .projects.plugin import setup_projects
 from .publications import setup_publications
 from .rabbitmq import setup_rabbitmq

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Literal, Optional
+from typing import Any, Literal
 
 from aiohttp import web
 from models_library.emails import LowerCaseEmailStr
@@ -13,7 +13,7 @@
     send_email_from_template,
 )
 from .login.decorators import login_required
-from .products import Product, get_current_product, get_product_template_path
+from .products.plugin import Product, get_current_product, get_product_template_path
 from .security_decorators import permission_required
 from .utils import get_traceback_string
 from .utils_aiohttp import envelope_json_response
@@ -27,7 +27,7 @@
 
 
 class TestEmail(BaseModel):
-    from_: Optional[LowerCaseEmailStr] = Field(None, description="Email sender")
+    from_: LowerCaseEmailStr | None = Field(None, description="Email sender")
     to: LowerCaseEmailStr = Field(..., description="Email receiver")
     template_name: Literal[
         "change_email_email.jinja2",

@@ -13,7 +13,8 @@
 
 from . import groups_handlers
 from ._constants import APP_OPENAPI_SPECS_KEY, APP_SETTINGS_KEY
-from .products import setup_products
+from .products.plugin import setup_products
+
 
 logger = logging.getLogger(__name__)
 

@@ -2,7 +2,6 @@
 import json
 import logging
 from contextlib import suppress
-from typing import Optional
 
 from aiohttp import web
 from models_library.emails import LowerCaseEmailStr
@@ -62,7 +61,7 @@ async def list_groups(request: web.Request):
 
     List of the groups I belonged to
     """
-    from .products import Product, get_current_product
+    from .products.plugin import Product, get_current_product
 
     product: Product = get_current_product(request)
     user_id = request[RQT_USERID_KEY]
@@ -285,7 +284,7 @@ async def get_scicrunch_resource(request: web.Request):
 
     # check if in database first
     repo = ResearchResourceRepository(request.app)
-    resource: Optional[ResearchResource] = await repo.get_resource(rrid)
+    resource: ResearchResource | None = await repo.get_resource(rrid)
     if not resource:
         # otherwise, request to scicrunch service
         scicrunch = SciCrunch.get_instance(request.app)
@@ -306,7 +305,7 @@ async def add_scicrunch_resource(request: web.Request):
 
     # check if exists
     repo = ResearchResourceRepository(request.app)
-    resource: Optional[ResearchResource] = await repo.get_resource(rrid)
+    resource: ResearchResource | None = await repo.get_resource(rrid)
     if not resource:
         # then request scicrunch service
         scicrunch = SciCrunch.get_instance(request.app)

@@ -10,7 +10,7 @@
 from servicelib.error_codes import create_error_code
 from servicelib.mimetype_constants import MIMETYPE_APPLICATION_JSON
 
-from ..products import Product, get_current_product
+from ..products.plugin import Product, get_current_product
 from ..session_access import session_access_required
 from ._2fa import (
     create_2fa_code,

@@ -13,7 +13,7 @@
 from simcore_postgres_database.models.users import UserRole
 
 from .._meta import API_VTAG
-from ..products import Product, get_current_product
+from ..products.plugin import Product, get_current_product
 from ..security_api import check_password, forget
 from ..session_access import on_success_grant_session_access_to, session_access_required
 from ..utils_aiohttp import NextPage

@@ -9,7 +9,7 @@
 from servicelib.request_keys import RQT_USERID_KEY
 
 from .._meta import API_VTAG
-from ..products import Product, get_current_product
+from ..products.plugin import Product, get_current_product
 from ..security_api import check_password, encrypt_password
 from ..utils import HOUR
 from ..utils_rate_limiting import global_rate_limit_route

@@ -13,7 +13,7 @@
 from simcore_postgres_database.errors import UniqueViolation
 from yarl import URL
 
-from ..products import Product, get_current_product
+from ..products.plugin import Product, get_current_product
 from ..security_api import encrypt_password
 from ..session_access import session_access_required
 from ..utils import MINUTE

@@ -1,6 +1,6 @@
 import logging
 from datetime import datetime, timedelta
-from typing import Literal, Optional
+from typing import Literal
 
 from aiohttp import web
 from aiohttp.web import RouteTableDef
@@ -13,7 +13,7 @@
 from .._meta import API_VTAG
 from ..groups_api import auto_add_user_to_groups, auto_add_user_to_product_group
 from ..invitations import is_service_invitation_code
-from ..products import Product, get_current_product
+from ..products.plugin import Product, get_current_product
 from ..security_api import encrypt_password
 from ..session_access import on_success_grant_session_access_to, session_access_required
 from ..utils import MINUTE
@@ -70,7 +70,7 @@ class InvitationCheck(InputSchema):
 
 
 class InvitationInfo(InputSchema):
-    email: Optional[LowerCaseEmailStr] = Field(
+    email: LowerCaseEmailStr | None = Field(
         None, description="Email associated to invitation or None"
     )
 
@@ -112,8 +112,8 @@ async def check_registration_invitation(request: web.Request):
 class RegisterBody(InputSchema):
     email: LowerCaseEmailStr
     password: SecretStr
-    confirm: Optional[SecretStr] = Field(None, description="Password confirmation")
-    invitation: Optional[str] = Field(None, description="Invitation code")
+    confirm: SecretStr | None = Field(None, description="Password confirmation")
+    invitation: str | None = Field(None, description="Invitation code")
 
     _password_confirm_match = validator("confirm", allow_reuse=True)(
         check_confirm_password_match
@@ -151,7 +151,7 @@ async def register(request: web.Request):
 
     await check_other_registrations(email=registration.email, db=db, cfg=cfg)
 
-    expires_at: Optional[datetime] = None  # = does not expire
+    expires_at: datetime | None = None  # = does not expire
     if settings.LOGIN_REGISTRATION_INVITATION_REQUIRED:
         # Only requests with INVITATION can register user
         # to either a permanent or to a trial account
@@ -257,7 +257,7 @@ class RegisterPhoneBody(InputSchema):
 
 
 class _PageParams(BaseModel):
-    retry_2fa_after: Optional[PositiveInt] = None
+    retry_2fa_after: PositiveInt | None = None
 
 
 class RegisterPhoneNextPage(NextPage[_PageParams]):

@@ -20,7 +20,7 @@
 from ..email import setup_email
 from ..email_settings import get_plugin_settings as get_email_plugin_settings
 from ..invitations import setup_invitations
-from ..products import ProductName, list_products, setup_products
+from ..products.plugin import ProductName, list_products, setup_products
 from ..redis import setup_redis
 from ..rest import setup_rest
 from ._constants import APP_LOGIN_SETTINGS_PER_PRODUCT_KEY

@@ -6,7 +6,7 @@
 
 from .._resources import resources
 from ..email_core import AttachmentTuple, send_email_from_template
-from ..products import get_product_template_path
+from ..products.plugin import get_product_template_path
 
 log = logging.getLogger(__name__)
 

@@ -1,14 +1,14 @@
 import logging
-from typing import AsyncIterator, Optional
+from typing import AsyncIterator
 
 import sqlalchemy as sa
 from aiopg.sa.connection import SAConnection
 from aiopg.sa.result import ResultProxy, RowProxy
 from simcore_postgres_database.models.products import jinja2_templates
 
-from .db_base_repository import BaseRepository
-from .db_models import products
-from .products_model import Product
+from ..db_base_repository import BaseRepository
+from ..db_models import products
+from ._model import Product
 
 log = logging.getLogger(__name__)
 
@@ -31,30 +31,31 @@ async def iter_products(conn: SAConnection) -> AsyncIterator[ResultProxy]:
 
 
 class ProductRepository(BaseRepository):
-    async def get_product(self, product_name: str) -> Optional[Product]:
+    async def get_product(self, product_name: str) -> Product | None:
         async with self.engine.acquire() as conn:
             result: ResultProxy = await conn.execute(
                 sa.select(_COLUMNS_IN_MODEL).where(products.c.name == product_name)
             )
-            row: Optional[RowProxy] = await result.first()
+            row: RowProxy | None = await result.first()
             return Product.from_orm(row) if row else None
 
     async def get_template_content(
         self,
         template_name: str,
-    ) -> Optional[str]:
+    ) -> str | None:
         async with self.engine.acquire() as conn:
-            return await conn.scalar(
+            template_content: str | None = await conn.scalar(
                 sa.select([jinja2_templates.c.content]).where(
                     jinja2_templates.c.name == template_name
                 )
             )
+            return template_content
 
     async def get_product_template_content(
         self,
         product_name: str,
         product_template: sa.Column = products.c.registration_email_template,
-    ) -> Optional[str]:
+    ) -> str | None:
         async with self.engine.acquire() as conn:
             oj = sa.join(
                 products,

@@ -11,10 +11,10 @@
     get_or_create_product_group,
 )
 
-from ._constants import APP_DB_ENGINE_KEY, APP_PRODUCTS_KEY
-from .products_db import iter_products
-from .products_model import Product
-from .statics_constants import FRONTEND_APP_DEFAULT, FRONTEND_APPS_AVAILABLE
+from .._constants import APP_DB_ENGINE_KEY, APP_PRODUCTS_KEY
+from ..statics_constants import FRONTEND_APP_DEFAULT, FRONTEND_APPS_AVAILABLE
+from ._db import iter_products
+from ._model import Product
 
 log = logging.getLogger(__name__)
 
@@ -76,7 +76,7 @@ async def load_products_on_startup(app: web.Application):
     async with engine.acquire() as conn:
         async for row in iter_products(conn):
             try:
-                name = row.name  # type:ignore
+                name = row.name
                 app_products[name] = Product.from_orm(row)
 
                 assert name in FRONTEND_APPS_AVAILABLE  # nosec

@@ -3,9 +3,9 @@
 from aiohttp import web
 from servicelib.aiohttp.typing_extension import Handler
 
-from ._constants import APP_PRODUCTS_KEY, RQ_PRODUCT_KEY, X_PRODUCT_NAME_HEADER
-from ._meta import API_VTAG
-from .products_model import Product
+from .._constants import APP_PRODUCTS_KEY, RQ_PRODUCT_KEY, X_PRODUCT_NAME_HEADER
+from .._meta import API_VTAG
+from ._model import Product
 
 log = logging.getLogger(__name__)
 

@@ -1,6 +1,6 @@
 import logging
 import string
-from typing import Any, Optional, Pattern, Union
+from typing import Any, Pattern
 
 from models_library.basic_regex import (
     PUBLIC_VARIABLE_NAME_RE,
@@ -20,8 +20,8 @@
     WebFeedback,
 )
 
-from .db_models import products
-from .statics_constants import FRONTEND_APPS_AVAILABLE
+from ..db_models import products
+from ..statics_constants import FRONTEND_APPS_AVAILABLE
 
 log = logging.getLogger(__name__)
 
@@ -42,7 +42,7 @@ class Product(BaseModel):
     name: ProductName = Field(regex=PUBLIC_VARIABLE_NAME_RE)
 
     display_name: str = Field(..., description="Long display name")
-    short_name: Optional[str] = Field(
+    short_name: str | None = Field(
         None,
         regex=TWILIO_ALPHANUMERIC_SENDER_ID_RE,
         min_length=2,
@@ -58,37 +58,37 @@ class Product(BaseModel):
         " Other support emails can be defined under 'support' field",
     )
 
-    twilio_messaging_sid: Optional[str] = Field(
+    twilio_messaging_sid: str | None = Field(
         default=None, min_length=34, max_length=34, description="Identifier for SMS"
     )
 
-    vendor: Optional[Vendor] = Field(
+    vendor: Vendor | None = Field(
         None,
         description="Vendor information such as company name, address, copyright, ...",
     )
 
-    issues: Optional[list[IssueTracker]] = None
+    issues: list[IssueTracker] | None = None
 
-    manuals: Optional[list[Manual]] = None
+    manuals: list[Manual] | None = None
 
-    support: Optional[list[Union[Forum, EmailFeedback, WebFeedback]]] = Field(None)
+    support: list[Forum | EmailFeedback | WebFeedback] | None = Field(None)
 
     login_settings: ProductLoginSettingsDict = Field(
         ...,
         description="Product customization of login settings. "
         "Note that these are NOT the final plugin settings but those are obtained from login.settings.get_plugin_settings",
     )
 
-    registration_email_template: Optional[str] = Field(
+    registration_email_template: str | None = Field(
         None, x_template_name="registration_email"
     )
 
-    max_open_studies_per_user: Optional[PositiveInt] = Field(
+    max_open_studies_per_user: PositiveInt | None = Field(
         default=None,
         description="Limits the number of studies a user may have open concurently (disabled if NULL)",
     )
 
-    group_id: Optional[int] = Field(
+    group_id: int | None = Field(
         default=None, description="Groups associated to this product"
     )
 
@@ -232,10 +232,11 @@ def to_statics(self) -> dict[str, Any]:
             by_alias=True,
         )
 
-    def get_template_name_for(self, filename: str) -> Optional[str]:
+    def get_template_name_for(self, filename: str) -> str | None:
         """Checks for field marked with 'x_template_name' that fits the argument"""
         template_name = filename.removesuffix(".jinja2")
         for field in self.__fields__.values():
             if field.field_info.extra.get("x_template_name") == template_name:
-                return getattr(self, field.name)
+                template_name_attribute: str = getattr(self, field.name)
+                return template_name_attribute
         return None