- fix CI

Author: manpace

app/__init__.py

@@ -16,12 +16,16 @@
 from app.utils.utils import remove_str_prefix, sanitize_for_logging
 
 DEBUG_LOG_FORMAT = "%(asctime)s.%(msecs)03d | %(short_logger_name)-10.15s | %(levelname)-8s | %(correlation_id)s | %(filename)s:%(lineno)d | %(message)s"
-LOG_FORMAT = "%(asctime)s.%(msecs)03d | %(short_logger_name)-10.15s | %(levelname)-8s | %(correlation_id)s | %(message)s"
+LOG_FORMAT = (
+    "%(asctime)s.%(msecs)03d | %(short_logger_name)-10.15s | %(levelname)-8s | %(correlation_id)s | %(message)s"
+)
+
 
 class PackageOnlyFilter(logging.Filter):
     """
     Logging filter that allows only log records originating from this application package.
     """
+
     def filter(self, record):
         return record.name.startswith(__name__)
 
@@ -36,17 +40,15 @@ class LogFormatFilter(logging.Filter):
 
     def filter(self, record):
         record.correlation_id = (
-            getattr(g, "correlation_id", CORRELATION_ID_FALLBACK)
-            if has_app_context()
-            else CORRELATION_ID_FALLBACK
+            getattr(g, "correlation_id", CORRELATION_ID_FALLBACK) if has_app_context() else CORRELATION_ID_FALLBACK
         )
-        record.short_logger_name = record.name.split(".")[0] #extract main module name
+        record.short_logger_name = record.name.split(".")[0]  # extract main module name
         return True
 
 
 def _loader_include(loader, node):
     nome_file = os.path.join(os.path.dirname(loader.name), loader.construct_scalar(node))
-    with open(nome_file, 'r', encoding='utf-8') as f:
+    with open(nome_file, "r", encoding="utf-8") as f:
         return yaml.load(f, Loader=type(loader))
 
 
@@ -60,13 +62,12 @@ def _loader_env_var(loader, node):
     raw_value = loader.construct_scalar(node)
     new_value = os.environ.get(raw_value)
     if new_value is None:
-        msg = "Cannot construct value from {node}: {value}".format(
-            node=node, value=new_value
-        )
+        msg = "Cannot construct value from {node}: {value}".format(node=node, value=new_value)
         raise yaml.YAMLError(msg)
     return new_value
 
-yaml.SafeLoader.add_constructor('!INCLUDE', _loader_include)
+
+yaml.SafeLoader.add_constructor("!INCLUDE", _loader_include)
 yaml.SafeLoader.add_constructor("!ENV", _loader_env_var)
 
 
@@ -82,28 +83,33 @@ def setup_logging(app):
         fmt = logging.Formatter(LOG_FORMAT, datefmt="%Y-%m-%d %H:%M:%S")
 
     if settings.filename and settings.filepath:
-        _handler = RotatingFileHandler(os.path.join(settings.filepath, settings.filename), maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8")
+        _handler = RotatingFileHandler(
+            os.path.join(settings.filepath, settings.filename),
+            maxBytes=10 * 1024 * 1024,
+            backupCount=5,
+            encoding="utf-8",
+        )
     else:
         _handler = logging.StreamHandler(sys.stdout)
 
     _handler.setLevel(_level)
     _handler.setFormatter(fmt)
     _handler.addFilter(LogFormatFilter())
 
-    #configure ROOT logger
+    # configure ROOT logger
     root_logger = logging.getLogger()
     root_logger.setLevel(_level)
     root_logger.handlers.clear()
     root_logger.addHandler(_handler)
 
-    app.logger.setLevel(_level) #flask
+    app.logger.setLevel(_level)  # flask
 
-    if not settings.libs_enabled: #disable lib logger
+    if not settings.libs_enabled:  # disable lib logger
         _handler.addFilter(PackageOnlyFilter())
     else:
-        logging.getLogger("werkzeug") #init a lazy logger otherwise not in loggerDict
+        logging.getLogger("werkzeug")  # init a lazy logger otherwise not in loggerDict
         for logger_name in logging.Logger.manager.loggerDict:
-            if logger_name.startswith(__name__): #skip app module
+            if logger_name.startswith(__name__):  # skip app module
                 continue
             logging.getLogger(logger_name).setLevel(settings.libs_level)
 
@@ -115,23 +121,21 @@ def load_config(app):
         raise ValueError(f"Failed to load configuration: The folder {config_path} does not exist.")
 
     try:
-        with open(os.path.join(config_path, "app_config.yaml"), 'r', encoding='utf-8') as f:
+        with open(os.path.join(config_path, "app_config.yaml"), "r", encoding="utf-8") as f:
             _config = yaml.safe_load(f)
             config = AppConfig.model_validate(_config)
             app.config.update(dict(SETTINGS=config))
 
             config_data = {
-                nome: getattr(config, nome)
-                for nome, field in config.model_fields.items()
-                if field.annotation is dict
+                nome: getattr(config, nome) for nome, field in config.model_fields.items() if field.annotation is dict
             }
             app.config.update(config_data)
     except Exception as e:
         raise ValueError("Failed to load configuration") from e
 
 
 def create_app():
-    #setup app
+    # setup app
     app = Flask(__name__)
     load_config(app)
     app.secret_key = app.config[APP_SETTINGS_KEY].app.secret_key

app/constants.py

@@ -52,9 +52,32 @@
 
 # EU member state ISO-3166-1 alpha-2 codes
 EU_COUNTRIES: set[str] = {
-    "AT", "BE", "BG", "CZ", "CY", "DK", "DE", "EE", "ES", "FR",
-    "FI", "GR", "HU", "IE", "IT", "LV", "LT", "LU", "MT", "NL",
-    "PL", "PT", "RO", "SK", "SI", "SE",
+    "AT",
+    "BE",
+    "BG",
+    "CZ",
+    "CY",
+    "DK",
+    "DE",
+    "EE",
+    "ES",
+    "FR",
+    "FI",
+    "GR",
+    "HU",
+    "IE",
+    "IT",
+    "LV",
+    "LT",
+    "LU",
+    "MT",
+    "NL",
+    "PL",
+    "PT",
+    "RO",
+    "SK",
+    "SI",
+    "SE",
 }
 
 # Valid IdP hint identifiers
@@ -70,4 +93,3 @@
 JWT_PREFIX: str = "jwt"
 SD_JWT_PREFIX: str = "dc_sd_jwt"
 MSO_MDOC_PREFIX: str = "mso_mdoc"
-

app/models/__init__.py

@@ -1,4 +1,3 @@
 from app.models.config.app_config import AppConfig
 
 __all__ = ["AppConfig"]
-

app/models/config/app_config.py

@@ -5,7 +5,8 @@
 from app.models.config.credentials_config import CredentialsConfig
 from app.models.config.provider_config import ProviderConfig
 
-#TODO: validations to be implemented
+# TODO: validations to be implemented
+
 
 class LogSetting(BaseModel):
     filepath: Optional[str] = None
@@ -14,6 +15,7 @@ class LogSetting(BaseModel):
     libs_enabled: bool
     libs_level: Optional[Literal["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"]] = "INFO"
 
+
 class AppSettings(BaseModel):
     secret_key: str
     logging: LogSetting
@@ -23,6 +25,7 @@ class AppSettings(BaseModel):
     favicon_subpath: str
     static_folder: str
 
+
 class AppConfig(BaseModel):
     provider_config: ProviderConfig
     credentials_config: CredentialsConfig

app/models/config/credentials_config.py

@@ -3,11 +3,12 @@
 from pydantic import BaseModel, Field
 
 
-#TODO: validations to be implemented
+# TODO: validations to be implemented
 class DocumentIdentifier(BaseModel):
     type: str
     value: str
 
+
 class DocumentFormat(BaseModel):
     id: str
     specs: dict[str, Any] | None = Field(default_factory=dict)
@@ -20,6 +21,7 @@ class Credential(BaseModel):
     document_identifier: DocumentIdentifier
     document_format: DocumentFormat
 
+
 class CredentialsConfig(BaseModel):
     supported_credentials: dict[str, Credential]
     internal_attributes_mappings: dict[str, Any]

app/models/config/provider_config.py

@@ -5,27 +5,23 @@
 
 
 class ProviderConfig(BaseModel):
-
-    #TODO: refactoring: remove model_config and computed_field -> create plain BaseModel + validations to be implemented
-    model_config = {"extra": "allow"} #temporary workaround
+    # TODO: refactoring: remove model_config and computed_field -> create plain BaseModel + validations to be implemented
+    model_config = {"extra": "allow"}  # temporary workaround
     _config: Dict[str, Any] = PrivateAttr()
 
-
     def __init__(self, **kwargs):
         if not kwargs:
             raise Exception("Invalid provider config")
         super().__init__(**kwargs)
         self._config = kwargs
 
-
     @model_validator(mode="before")
     @classmethod
     def check_config_present(cls, data: Any) -> Any:
         if data is None:
             raise Exception("Invalid provider config")
         return data
 
-
     @computed_field
     @property
     def spec_version(self) -> str:

app/routes/main_routes.py

@@ -7,6 +7,7 @@
 
 main_routes = Blueprint("main_routes", __name__)
 
+
 @main_routes.app_errorhandler(404)
 def page_not_found(e):
     return render_template("404.html"), 404

app/routes/wallet_routes.py

@@ -56,6 +56,7 @@ def split_string(value, delimiter):
         return []
     return str(value).split(delimiter)
 
+
 @wallet_routes.app_template_filter("format_date")
 def format_date(value, fmt="%d-%m-%Y %H:%M:%S"):
     """
@@ -66,6 +67,7 @@ def format_date(value, fmt="%d-%m-%Y %H:%M:%S"):
     Returns str(value) if unparseable, empty string if None.
     """
     from datetime import timezone
+
     if value is None:
         return ""
     try:
@@ -368,7 +370,9 @@ def credentialTypeTemplate():
     status_descr = "VALIDO"
 
     data_row = response.get("data_row", {})
-    internal_claims = parser_credential_format_to_internal(current_app.config[APP_SETTINGS_KEY].credentials_config, key.rsplit("_", 1)[0], claims)
+    internal_claims = parser_credential_format_to_internal(
+        current_app.config[APP_SETTINGS_KEY].credentials_config, key.rsplit("_", 1)[0], claims
+    )
 
     # @TODO Decprecated
     content_list = []
@@ -414,19 +418,15 @@ def presentation_phase():
     it_wallet_service = ItWalletService(session, external_discovery=True)
     try:
         params = parse_url_to_dict(qrcode_data)
-        credentials = it_wallet_service.loginToVerifier(params.get("client_id"),params.get("request_uri"),params.get("request_uri_method"), params.get("state"))
+        credentials = it_wallet_service.loginToVerifier(
+            params.get("client_id"), params.get("request_uri"), params.get("request_uri_method"), params.get("state")
+        )
     except Exception as e:
         logger.error(f"Error, message: {e}")
         # @Todo insert this into Util
-        return jsonify({
-            "success": False,
-            "error_message": f"Exception :{e}."
-        }), 400
-    return jsonify({
-            "success": True,
-            "success_message": "Login success.",
-            "data": credentials
-        }), 200
+        return jsonify({"success": False, "error_message": f"Exception :{e}."}), 400
+    return jsonify({"success": True, "success_message": "Login success.", "data": credentials}), 200
+
 
 @wallet_routes.route("/authorization", methods=["POST"], strict_slashes=False)
 def authorization_phase():
@@ -441,10 +441,8 @@ def authorization_phase():
     except Exception as e:
         logger.error(f"Error, message: {e}")
         # @Todo insert this into Util
-        return jsonify({
-            "success": False,
-            "error_message": f"Exception :{e}."
-        }), 400
+        return jsonify({"success": False, "error_message": f"Exception :{e}."}), 400
+
 
 def _get_template_name_for_credential_key(key: str) -> str | None:
     """

app/service/itwallet_helpers.py

@@ -189,7 +189,11 @@ def _validate_ec_iss_sub(ec_payload: dict, expected: str) -> None:
 
 def _validate_ec_authority_hints(ec_payload: dict, expected_hint: Any) -> None:
     """Validate EC authority_hints contains expected_hint. Raises ValueError."""
-    logger.debug(f"Entering method: _validate_ec_authority_hints. Params [ec_payload: {ec_payload}, expected_hint: {expected_hint}]", ec_payload, expected_hint)
+    logger.debug(
+        f"Entering method: _validate_ec_authority_hints. Params [ec_payload: {ec_payload}, expected_hint: {expected_hint}]",
+        ec_payload,
+        expected_hint,
+    )
     hints = ec_payload.get("authority_hints", [])
     if not isinstance(hints, list) or not hints or expected_hint not in hints:
         raise ValueError(f"EC 'authority_hints' non valido o mancante hint '{expected_hint}'")
@@ -198,7 +202,10 @@ def _validate_ec_authority_hints(ec_payload: dict, expected_hint: Any) -> None:
 def _validate_ec_metadata_and_jwks(ec_payload: dict, expected_metadata_types: list) -> None:
     """Validate EC has required metadata types and jwks. Raises ValueError."""
     from app.constants import METADATA_TYPE_FEDERATION_ENTITY
-    logger.debug(f"Entering method: _validate_ec_metadata_and_jwks. Params [ec_payload: {ec_payload}, expected_metadata_types: {expected_metadata_types}]")
+
+    logger.debug(
+        f"Entering method: _validate_ec_metadata_and_jwks. Params [ec_payload: {ec_payload}, expected_metadata_types: {expected_metadata_types}]"
+    )
     actual = ec_payload.get("metadata", {})
     missing = [t for t in expected_metadata_types if t not in actual]
     if missing:
@@ -216,7 +223,9 @@ def validate_ec(
     ec_payload: dict, expected_issuer_url: str, expected_metadata_types: list, expected_hint: Any = None
 ) -> None:
     """Validate Entity Configuration: iss/sub, authority_hints, metadata types, jwks. Raises ValueError."""
-    logger.info(f"Entering method: validate_ec. Params [ec_payload: {ec_payload}, expected_issuer_url: {expected_issuer_url}, expected_metadata_types: {expected_metadata_types}, expected_hint: {expected_hint}]")
+    logger.info(
+        f"Entering method: validate_ec. Params [ec_payload: {ec_payload}, expected_issuer_url: {expected_issuer_url}, expected_metadata_types: {expected_metadata_types}, expected_hint: {expected_hint}]"
+    )
     if not ec_payload:
         raise ValueError("Entity Configuration non specificato")
     _validate_ec_iss_sub(ec_payload, expected_issuer_url)