1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

from pydantic import BaseModel, EmailStr, Field, field_validator, root_validator, ValidationError
from typing import Optional, NewType, Any, TypeAlias, Literal, Union
from PIL import Image, JpegImagePlugin, PngImagePlugin
import base64
import io
import json
from enum import Enum
import numpy.typing as npt
import numpy as np

class XXXEntity(BaseModel):
    pass

class MicroserviceInferenceRepresentationDTOV1(BaseModel):
    version: Literal[1]
    xxx: list[XXXEntity]

class MicroserviceInferenceRepresentationDTOV2(BaseModel):
    version: Literal[2]
    xxx: list[XXXEntity]
    zzz: list[XXXEntity] | None = None

MicroserviceInferenceRepresentationDTO = MicroserviceInferenceRepresentationDTOV1 | MicroserviceInferenceRepresentationDTOV2


def parse_microservice_inference_representation(body: dict[str, Any]) -> MicroserviceInferenceRepresentationDTO:
    version = body.get("version", -1)
    if version == 1:
        return MicroserviceInferenceRepresentationDTOV1.parse_obj(body)
    elif version == 2:
        return MicroserviceInferenceRepresentationDTOV2.parse_obj(body)
    else:
        raise ValueError("Unsupported version provided")

# 使用例
body = {
    "version": 2,
    "xxx": [],
}

try:
    dto = parse_microservice_inference_representation(body)
    print(f"DTO parsed successfully: {type(dto)}")
except (ValidationError, ValueError) as e:
    print(f"Error parsing DTO: {e}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator
from typing import Optional, NewType, Any, TypeAlias
from PIL import Image, JpegImagePlugin, PngImagePlugin
import base64
import io
import json
from enum import Enum
import numpy.typing as npt
import numpy as np

UserId = NewType('UserId', int)
CustomerAnalysisId = NewType('CustomerAnalysisId', int)

ImageNp: TypeAlias = npt.NDArray[np.uint8]
ImagePil: TypeAlias = Image.Image

class ImageUtil:
    @staticmethod
    def base64_to_image(base64_str: str):
        image_data = base64.b64decode(base64_str)
        image = Image.open(io.BytesIO(image_data))
        return image

    @staticmethod
    def image_to_byte_array(image: ImagePil):
        img_byte_arr = io.BytesIO()
        image_format = 'JPEG' if isinstance(image, JpegImagePlugin.JpegImageFile) else 'PNG'
        image.save(img_byte_arr, format=image_format)
        return img_byte_arr.getvalue()

    @classmethod
    def image_to_base64_str(cls, image: ImagePil):
        return base64.b64encode(cls.image_to_byte_array(image)).decode('utf-8')


class MyBaseModel(BaseModel):
    class Config:
        # デフォルトでは、Pydanticは標準のデータ型しか型が使えないため
        # 例えば、int、str、float、list、dict など）およびPydanticモデルのインスタンス
        arbitrary_types_allowed = True

class UserProfileCategory(str,Enum):
    STANDARD = "standard"
    ADMIN = "admin"
    GUEST = "guest"

class UsernameValue(BaseModel):
    value: str

    @field_validator('value')
    def validate_username(cls, v):
        if not v.isalnum() or len(v) < 3:
            raise ValueError('Username must be at least 3 characters long and contain only alphanumeric characters')
        return v

class EmailValue(BaseModel):
    value: EmailStr

class ProfileImageValue(MyBaseModel):
    value: ImagePil
    mime_type: str

    class Config:
        json_encoders = {
            ImagePil: ImageUtil.image_to_base64_str,
        }

class CustomerAnalysisEntity(BaseModel):
    id: CustomerAnalysisId
    user_id: UserId
    username: UsernameValue | None = None
    email: EmailValue | None = None
    profile_image: ProfileImageValue | None = None
    user_type: UserProfileCategory

    @model_validator(mode="after")
    def validate_root(cls, values):
        if not (values.username or values.email):
            raise ValueError('Either username or email must be provided.')
        return values

    def to_json(self):
        return self.json(exclude_none=True)

    def to_dict(self):
        return self.dict(exclude={})

    @classmethod
    def from_dict(cls, data: dict[str, Any]):
        if 'profile_image' in data and data['profile_image'] is not None and 'value' in data['profile_image']:
            base64_str = data['profile_image']['value']
            image = ImageUtil.base64_to_image(base64_str)
            data['profile_image']['value'] = image
        return cls(**data)

    @classmethod
    def from_json(cls, data: str):
        json_data = json.loads(data) # str -> dict
        return cls.from_dict(json_data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

class Test:
    img_path = 'image.png'
    @classmethod
    def to_json(cls):
        img = Image.open(cls.img_path)
        customer_analysis = CustomerAnalysisEntity(
            id=CustomerAnalysisId(111),
            user_id=UserId(12345),
            username=UsernameValue(value="JohnDoe"),
            email=EmailValue(value="johndoe@example.com"),
            profile_image=ProfileImageValue(
                value=img,
                mime_type="image/jpeg"
            ),
            user_type=UserProfileCategory.STANDARD
        )

        user_profile_json = customer_analysis.json()
        file_path = 'customer_analysis.json'
        with open(file_path, 'w') as file:
            file.write(user_profile_json)

    @staticmethod
    def from_json():
        file_path = 'customer_analysis.json'
        with open(file_path, 'r') as file:
            user_profile_json = file.read()
        customer_analysis = CustomerAnalysisEntity.from_json(user_profile_json)
        print(customer_analysis)

    @staticmethod
    def to_dict():
        file_path = 'customer_analysis.json'
        with open(file_path, 'r') as file:
            user_profile_json = file.read()
        customer_analysis = CustomerAnalysisEntity.from_json(user_profile_json)
        user_json = customer_analysis.to_dict()
        print(user_json)

    @staticmethod
    def from_dict():
        user_profile_dict = {
            "id": 111,
            "user_id": 12345,
            "username": {"value": "JohnDoe"},
            "email": {"value": "johndoe@example.com"},
            "profile_image": None,
            "user_type": "standard"
        }
        customer_analysis = CustomerAnalysisEntity.from_dict(user_profile_dict)
        print(customer_analysis)



if __name__ == "__main__":
    Test.to_json()
    Test.to_dict()
    Test.from_json()
    Test.from_dict()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

class XXXInputDTO(BaseModel):
    aaa: str
    bbb: str

class XXXOutputDTO(BaseModel):
    xxx: str
    yyy: str

def domain_subdomain1_xxx_serveice(
  xxx_input_dto: XXXInputDTO, 
  xxx_repo: IXXRepository, 
  yyy_repo: IyyyRepository
) -> XXXOutputDTO:
  pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

$ tree -L 3
└── common
    ├── models
    │   ├── representation_dto.py
└── domain
    ├── subdomain1
    │   ├── model
    │   │   ├── entity.py
    │   │   ├── value_object.py
    │   ├── service.py
    │   ├── repository_interface.py
    ├── subdomain2
    │   ├── model
    │   │   ├── entity.py
    │   │   ├── value_object.py
    │   ├── service.py
    │   ├── repository_interface.py
└── application
    ├── use_cases
    │   ├── subdomain1_use_case.py
    │   ├── subdomain2_use_case.py
└── infrastructure
    ├── repository
    │   ├── subdomain1_repository_impl.py
    │   ├── subdomain2_repository_impl.py
    │   ├── dao
    │   │   ├── xxx_dao.py
    ├── external_services
    │   ├── xxx_service_adapter.py
    ├── database
    │   ├── orm_models.py
    │   ├── database_accessor.py
└── presentation
    ├── web
    │   ├── controllers
    │   ├── views
    │   ├── viewmodels
    ├── api
    │   ├── controllers
    │   ├── dtos

1
2
3
4
5
6
7
8
9

# app/use_cases/subdomain1_use_case.py
from domain.subdomain1.service import SomeDomainService
from infrastructure.repository.subdomain1_repository_impl import Subdomain1RepositoryImpl

def execute_some_use_case():
    repository = Subdomain1RepositoryImpl()
    domain_service = SomeDomainService(repository)
    result = domain_service.perform_some_operation()
    return result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

$ cd aaa_lib
$ tree -L 3
└── xxx_pj
    ├── aaa_lib
    │   ├── __init__.py
    │   ├── main.py

$ cd bbb_lib
$ tree -L 3
└── xxx_pj
    ├── bbb_lib
    │   ├── __init__.py
    │   ├── model
    │   │   ├── __init__.py
    │   │   ├── main.py