(PC-32146)[API] refactor: regroup ubble serialization pydantic models

api/src/pcapi/connectors/beneficiaries/ubble.py

@@ -11,7 +11,6 @@
 from pcapi.connectors.serialization import ubble_serializers
 from pcapi.core import logging as core_logging
 from pcapi.core.fraud import models as fraud_models
-from pcapi.core.fraud.ubble import models as ubble_fraud_models
 from pcapi.core.users import models as users_models
 from pcapi.models.feature import FeatureToggle
 from pcapi.utils import requests
@@ -21,9 +20,9 @@
 
 
 INCLUDED_MODELS = {
-    "documents": ubble_fraud_models.UbbleIdentificationDocuments,
-    "document-checks": ubble_fraud_models.UbbleIdentificationDocumentChecks,
-    "reference-data-checks": ubble_fraud_models.UbbleIdentificationReferenceDataChecks,
+    "documents": ubble_serializers.UbbleIdentificationDocuments,
+    "document-checks": ubble_serializers.UbbleIdentificationDocumentChecks,
+    "reference-data-checks": ubble_serializers.UbbleIdentificationReferenceDataChecks,
 }
 
 
@@ -173,7 +172,7 @@ def start_identification(  # pylint: disable=too-many-positional-arguments
         )
         response.raise_for_status()
 
-        ubble_identification = parse_obj_as(ubble_serializers.UbbleIdentificationResponse, response.json())
+        ubble_identification = parse_obj_as(ubble_serializers.UbbleV2IdentificationResponse, response.json())
         ubble_content = ubble_serializers.convert_identification_to_ubble_content(ubble_identification)
 
         logger.info(
@@ -374,7 +373,7 @@ def build_url(path: str, id_: int | str | None = None) -> str:
     return base_url + path
 
 
-def _get_included_attributes(response: dict, type_: str) -> ubble_fraud_models.UbbleIdentificationObject | None:
+def _get_included_attributes(response: dict, type_: str) -> ubble_serializers.UbbleIdentificationObject | None:
     filtered = [incl for incl in response["included"] if incl["type"] == type_]
     if not filtered:
         return None
@@ -419,13 +418,13 @@ def _extract_useful_content_from_response(
     response: dict,
 ) -> fraud_models.UbbleContent:
     documents = typing.cast(
-        ubble_fraud_models.UbbleIdentificationDocuments, _get_included_attributes(response, "documents")
+        ubble_serializers.UbbleIdentificationDocuments, _get_included_attributes(response, "documents")
     )
     document_checks = typing.cast(
-        ubble_fraud_models.UbbleIdentificationDocumentChecks, _get_included_attributes(response, "document-checks")
+        ubble_serializers.UbbleIdentificationDocumentChecks, _get_included_attributes(response, "document-checks")
     )
     reference_data_checks = typing.cast(
-        ubble_fraud_models.UbbleIdentificationReferenceDataChecks,
+        ubble_serializers.UbbleIdentificationReferenceDataChecks,
         _get_included_attributes(response, "reference-data-checks"),
     )
 

api/src/pcapi/connectors/serialization/ubble_serializers.py

@@ -1,13 +1,31 @@
 import contextlib
 import datetime
+import enum
 
 import pydantic.v1 as pydantic_v1
 
 from pcapi.core.fraud import models as fraud_models
-from pcapi.core.fraud.ubble import models as ubble_models
 from pcapi.core.users import models as users_models
 
 
+class UbbleIdentificationStatus(enum.Enum):
+    # ubble v2
+    PENDING = "pending"
+    CAPTURE_IN_PROGRESS = "capture_in_progress"
+    CHECKS_IN_PROGRESS = "checks_in_progress"
+    APPROVED = "approved"
+    DECLINED = "declined"
+    RETRY_REQUIRED = "retry_required"
+    REFUSED = "refused"
+    # ubble v1
+    UNINITIATED = "uninitiated"  # Identification has only been created (user has not started the verification flow)
+    INITIATED = "initiated"  # User has started the verification flow
+    PROCESSING = "processing"  # User has ended the verification flow, identification-url is not usable anymore
+    PROCESSED = "processed"  # Identification is completely processed by Ubble
+    ABORTED = "aborted"  # User has left the identification, the identification-url is no longer usable (this status is in beta test)
+    EXPIRED = "expired"  # The identification-url has expired and is no longer usable (only uninitiated and initiated identifications can become expired)
+
+
 class UbbleDeclaredData(pydantic_v1.BaseModel):
     name: str
     birth_date: datetime.date | None
@@ -45,13 +63,12 @@ class UbbleResponseCode(pydantic_v1.BaseModel):
     response_code: int
 
 
-class UbbleIdentificationResponse(pydantic_v1.BaseModel):
+class UbbleV2IdentificationResponse(pydantic_v1.BaseModel):
     # https://docs.ubble.ai/#tag/Identity-verifications/operation/create_and_start_identity_verification
     id: str
     applicant_id: str
     user_journey_id: str
-    # TODO clean up imports
-    status: ubble_models.UbbleIdentificationStatus
+    status: UbbleIdentificationStatus
     declared_data: UbbleDeclaredData
     links: UbbleLinks = pydantic_v1.Field(alias="_links")
     documents: list[UbbleDocument]
@@ -66,7 +83,7 @@ def document(self) -> UbbleDocument | None:
         return self.documents[0] if self.documents else None
 
     @property
-    def fraud_reason_codes(self) -> list[fraud_models.FraudReasonCode]:
+    def fraud_reason_codes(self) -> list["fraud_models.FraudReasonCode"]:
         return [
             fraud_models.UBBLE_REASON_CODE_MAPPING.get(
                 response_code.response_code, fraud_models.FraudReasonCode.ID_CHECK_BLOCKED_OTHER
@@ -78,7 +95,9 @@ class Config:
         use_enum_values = True
 
 
-def convert_identification_to_ubble_content(identification: UbbleIdentificationResponse) -> fraud_models.UbbleContent:
+def convert_identification_to_ubble_content(
+    identification: UbbleV2IdentificationResponse,
+) -> "fraud_models.UbbleContent":
     document = identification.document
     if not document:
         first_name, last_name = None, None
@@ -110,3 +129,137 @@ def convert_identification_to_ubble_content(identification: UbbleIdentificationR
         supported=None,
     )
     return content
+
+
+# DEPRECATED Ubble V1
+
+
+class UbbleScore(enum.Enum):
+    VALID = 1.0
+    INVALID = 0.0
+    UNDECIDABLE = -1.0
+
+
+class UbbleIdentificationObject(pydantic_v1.BaseModel):
+    # Parent class for any object defined in https://ubbleai.github.io/developer-documentation/#objects-2
+    pass
+
+
+class UbbleIdentificationAttributes(UbbleIdentificationObject):
+    # https://ubbleai.github.io/developer-documentation/#identifications
+    comment: str | None
+    created_at: datetime.datetime = pydantic_v1.Field(alias="created-at")
+    ended_at: datetime.datetime | None = pydantic_v1.Field(None, alias="ended-at")
+    identification_id: str = pydantic_v1.Field(alias="identification-id")
+    identification_url: str = pydantic_v1.Field(alias="identification-url")
+    number_of_attempts: int = pydantic_v1.Field(alias="number-of-attempts")
+    redirect_url: str = pydantic_v1.Field(alias="redirect-url")
+    score: float | None
+    started_at: datetime.datetime | None = pydantic_v1.Field(None, alias="started-at")
+    status: UbbleIdentificationStatus
+    status_updated_at: datetime.datetime = pydantic_v1.Field(alias="status-updated-at")
+    updated_at: datetime.datetime = pydantic_v1.Field(alias="updated-at")
+    user_agent: str | None = pydantic_v1.Field(None, alias="user-agent")
+    user_ip_address: str | None = pydantic_v1.Field(None, alias="user-ip-address")
+    webhook: str
+
+
+class UbbleReasonCode(UbbleIdentificationObject):
+    type: str = pydantic_v1.Field(alias="type")
+    id: int = pydantic_v1.Field(alias="id")
+
+
+class UbbleReasonCodes(UbbleIdentificationObject):
+    data: list[UbbleReasonCode]
+
+
+class UbbleIdentificationRelationships(UbbleIdentificationObject):
+    reason_codes: UbbleReasonCodes = pydantic_v1.Field(alias="reason-codes")
+
+
+class UbbleIdentificationData(pydantic_v1.BaseModel):
+    type: str
+    id: int
+    attributes: UbbleIdentificationAttributes
+    relationships: UbbleIdentificationRelationships
+
+
+class UbbleIdentificationDocuments(UbbleIdentificationObject):
+    # https://ubbleai.github.io/developer-documentation/#documents
+    birth_date: str | None = pydantic_v1.Field(None, alias="birth-date")
+    document_number: str | None = pydantic_v1.Field(None, alias="document-number")
+    document_type: str | None = pydantic_v1.Field(None, alias="document-type")
+    first_name: str | None = pydantic_v1.Field(None, alias="first-name")
+    gender: str | None = pydantic_v1.Field(None)
+    last_name: str | None = pydantic_v1.Field(None, alias="last-name")
+    married_name: str | None = pydantic_v1.Field(None, alias="married-name")
+    signed_image_front_url: str | None = pydantic_v1.Field(None, alias="signed-image-front-url")
+    signed_image_back_url: str | None = pydantic_v1.Field(None, alias="signed-image-back-url")
+
+
+class UbbleIdentificationDocumentChecks(UbbleIdentificationObject):
+    # https://ubbleai.github.io/developer-documentation/#document-checks
+    data_extracted_score: float | None = pydantic_v1.Field(None, alias="data-extracted-score")
+    expiry_date_score: float | None = pydantic_v1.Field(None, alias="expiry-date-score")
+    issue_date_score: float | None = pydantic_v1.Field(None, alias="issue-date-score")
+    live_video_capture_score: float | None = pydantic_v1.Field(None, alias="live-video-capture-score")
+    mrz_validity_score: float | None = pydantic_v1.Field(None, alias="mrz-validity-score")
+    mrz_viz_score: float | None = pydantic_v1.Field(None, alias="mrz-viz-score")
+    ove_back_score: float | None = pydantic_v1.Field(None, alias="ove-back-score")
+    ove_front_score: float | None = pydantic_v1.Field(None, alias="ove-front-score")
+    ove_score: float | None = pydantic_v1.Field(None, alias="ove-score")
+    quality_score: float | None = pydantic_v1.Field(None, alias="quality-score")
+    score: float | None = pydantic_v1.Field(None, alias="score")
+    supported: float | None = None
+    visual_back_score: float | None = pydantic_v1.Field(None, alias="visual-back-score")
+    visual_front_score: float | None = pydantic_v1.Field(None, alias="visual-front-score")
+
+
+class UbbleIdentificationFaceChecks(UbbleIdentificationObject):
+    # https://ubbleai.github.io/developer-documentation/#face-checks
+    active_liveness_score: float | None = pydantic_v1.Field(None, alias="active-liveness-score")
+    live_video_capture_score: float | None = pydantic_v1.Field(None, alias="live-video-capture-score")
+    quality_score: float | None = pydantic_v1.Field(None, alias="quality-score")
+    score: float | None = None
+
+
+class UbbleIdentificationReferenceDataChecks(UbbleIdentificationObject):
+    # https://ubbleai.github.io/developer-documentation/#reference-data-check
+    score: float | None = None
+
+
+class UbbleIdentificationDocFaceMatches(UbbleIdentificationObject):
+    # https://ubbleai.github.io/developer-documentation/#doc-face-matches
+    score: float | None = None
+
+
+class UbbleIdentificationIncluded(pydantic_v1.BaseModel):
+    type: str
+    id: int
+    attributes: UbbleIdentificationObject
+    relationships: dict | None
+
+
+class UbbleIdentificationIncludedDocuments(UbbleIdentificationIncluded):
+    attributes: UbbleIdentificationDocuments
+
+
+class UbbleIdentificationIncludedDocumentChecks(UbbleIdentificationIncluded):
+    attributes: UbbleIdentificationDocumentChecks
+
+
+class UbbleIdentificationIncludedFaceChecks(UbbleIdentificationIncluded):
+    attributes: UbbleIdentificationFaceChecks
+
+
+class UbbleIdentificationIncludedReferenceDataChecks(UbbleIdentificationIncluded):
+    attributes: UbbleIdentificationReferenceDataChecks
+
+
+class UbbleIdentificationIncludedDocFaceMatches(UbbleIdentificationIncluded):
+    attributes: UbbleIdentificationDocFaceMatches
+
+
+class UbbleIdentificationResponse(pydantic_v1.BaseModel):
+    data: UbbleIdentificationData
+    included: list[UbbleIdentificationIncluded]

api/src/pcapi/core/fraud/models.py

@@ -11,6 +11,7 @@
 from sqlalchemy.dialects import postgresql
 import sqlalchemy.orm as sa_orm
 
+from pcapi.connectors.serialization import ubble_serializers
 from pcapi.core.users import constants as users_constants
 from pcapi.core.users import models as users_models
 from pcapi.models import Base
@@ -19,7 +20,6 @@
 from pcapi.serialization.utils import to_camel
 
 from .common import models as common_models
-from .ubble import models as ubble_fraud_models
 
 
 if TYPE_CHECKING:
@@ -365,7 +365,7 @@ class UbbleContent(common_models.IdentityCheckContent):
     registration_datetime: datetime.datetime | None
     processed_datetime: datetime.datetime | None
     score: float | None
-    status: ubble_fraud_models.UbbleIdentificationStatus | None
+    status: ubble_serializers.UbbleIdentificationStatus | None
     status_updated_at: datetime.datetime | None
     supported: float | None
     signed_image_front_url: pydantic_v1.HttpUrl | None

api/src/pcapi/core/fraud/ubble/api.py

@@ -2,9 +2,9 @@
 import re
 
 from pcapi import settings
+from pcapi.connectors.serialization import ubble_serializers
 from pcapi.core.fraud import api as fraud_api
 from pcapi.core.fraud import models as fraud_models
-from pcapi.core.fraud.ubble import models as ubble_fraud_models
 from pcapi.core.subscription import api as subscription_api
 from pcapi.core.subscription.ubble import models as ubble_subsciption_models
 from pcapi.core.users import constants as users_constants
@@ -22,7 +22,7 @@ def on_ubble_result(fraud_check: fraud_models.BeneficiaryFraudCheck) -> None:
 
 
 def _ubble_readable_score(score: float | None) -> str:
-    return ubble_fraud_models.UbbleScore(score).name if score is not None else "AUCUN"
+    return ubble_serializers.UbbleScore(score).name if score is not None else "AUCUN"
 
 
 def _ubble_message_from_code(code: fraud_models.FraudReasonCode) -> str:
@@ -42,7 +42,7 @@ def _ubble_result_fraud_item(user: users_models.User, content: fraud_models.Ubbl
         status = fraud_models.FraudStatus.SUSPICIOUS
 
     # Decision from identification/score
-    if content.score == ubble_fraud_models.UbbleScore.VALID.value:
+    if content.score == ubble_serializers.UbbleScore.VALID.value:
         id_provider_detected_eligibility = subscription_api.get_id_provider_detected_eligibility(user, content)
         if id_provider_detected_eligibility:
             status = fraud_models.FraudStatus.OK
@@ -58,14 +58,14 @@ def _ubble_result_fraud_item(user: users_models.User, content: fraud_models.Ubbl
             elif age > users_constants.ELIGIBILITY_AGE_18:
                 reason_codes.add(fraud_models.FraudReasonCode.AGE_TOO_OLD)
                 detail = _ubble_message_from_code(fraud_models.FraudReasonCode.AGE_TOO_OLD).format(age=age)
-    elif content.score == ubble_fraud_models.UbbleScore.INVALID.value:
+    elif content.score == ubble_serializers.UbbleScore.INVALID.value:
         for score, reason_code in [
             (content.reference_data_check_score, fraud_models.FraudReasonCode.ID_CHECK_DATA_MATCH),
             (content.supported, fraud_models.FraudReasonCode.ID_CHECK_NOT_SUPPORTED),
             (content.expiry_date_score, fraud_models.FraudReasonCode.ID_CHECK_EXPIRED),
             (content.ove_score, fraud_models.FraudReasonCode.ID_CHECK_NOT_AUTHENTIC),
         ]:
-            if score == ubble_fraud_models.UbbleScore.INVALID.value:
+            if score == ubble_serializers.UbbleScore.INVALID.value:
                 status = fraud_models.FraudStatus.SUSPICIOUS
                 reason_codes.add(reason_code)
 
@@ -77,7 +77,7 @@ def _ubble_result_fraud_item(user: users_models.User, content: fraud_models.Ubbl
                 f" | document supported {_ubble_readable_score(content.supported)}"
                 f" | expiry-date-score {_ubble_readable_score(content.expiry_date_score)}"
             )
-    elif content.score == ubble_fraud_models.UbbleScore.UNDECIDABLE.value:
+    elif content.score == ubble_serializers.UbbleScore.UNDECIDABLE.value:
         status = fraud_models.FraudStatus.SUSPICIOUS
         # Add UNPROCESSABLE only if there are no other reason codes that would be more accurate
         if not reason_codes: