Update token and company list schemas

.dockerignore

@@ -11,4 +11,5 @@ docs/
 .DS_Store
 .cov/
 .venv
-Dockerfile
+Dockerfile
+.env

app/config.py

@@ -21,8 +21,15 @@
 import os
 from email.policy import HTTP, SMTP, SMTPUTF8
 
+from dotenv import load_dotenv
+
 BRAND_NAME = "ibet-Wallet-API"
 
+######################################################
+# Environment Setup
+######################################################
+load_dotenv()
+
 ####################################################
 # Basic settings
 ####################################################

app/model/db/company.py

@@ -19,7 +19,7 @@
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, Index, String, Text
+from sqlalchemy import CheckConstraint, DateTime, Index, String, Text
 from sqlalchemy.dialects.mysql import DATETIME as MySQLDATETIME
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -40,6 +40,12 @@ class Company(Base):
     rsa_publickey: Mapped[str | None] = mapped_column(String(2000))
     # Homepage URL
     homepage: Mapped[str | None] = mapped_column(Text)
+    # Trustee Corporate Name
+    trustee_corporate_name: Mapped[str | None] = mapped_column(String(30))
+    # Trustee Corporate Number
+    trustee_corporate_number: Mapped[str | None] = mapped_column(String(20))
+    # Trustee Corporate Address
+    trustee_corporate_address: Mapped[str | None] = mapped_column(String(60))
 
     if engine.name == "mysql":
         # NOTE:MySQLではDatetime型で小数秒桁を指定しない場合、整数秒しか保存されない
@@ -51,6 +57,11 @@ class Company(Base):
             DateTime, default=naive_utcnow, index=True
         )
     __table_args__ = (
+        CheckConstraint(
+            "((trustee_corporate_name IS NULL AND trustee_corporate_number IS NULL AND trustee_corporate_address IS NULL) "
+            "OR (trustee_corporate_name IS NOT NULL AND trustee_corporate_number IS NOT NULL AND trustee_corporate_address IS NOT NULL))",
+            name="ck_company_trustee_fields_complete",
+        ),
         # Covering Index
         Index(
             "ix_company_covering",
@@ -73,6 +84,13 @@ def json(self):
             "corporate_name": self.corporate_name,
             "rsa_publickey": self.rsa_publickey,
             "homepage": self.homepage,
+            "trustee": {
+                "corporate_name": self.trustee_corporate_name,
+                "corporate_number": self.trustee_corporate_number,
+                "corporate_address": self.trustee_corporate_address,
+            }
+            if self.trustee_corporate_name
+            else None,
         }
 
     FIELDS = {

app/model/db/public_info.py

@@ -49,16 +49,17 @@ class TokenList(Base):
     # Key Manager
     key_manager: Mapped[list[str]] = mapped_column(JSON, nullable=False)
     # Product Type
-    product_type: Mapped[Literal[1, 2, 3, 4, 5]] = mapped_column(
-        Integer, nullable=False
-    )
+    product_type: Mapped[int] = mapped_column(Integer, nullable=False)
+    # Issuer Address
+    issuer_address: Mapped[str | None] = mapped_column(String(42), nullable=True)
 
     def json(self):
         return {
             "token_address": self.token_address,
             "token_template": self.token_template,
             "key_manager": self.key_manager,
             "product_type": self.product_type,
+            "issuer_address": self.issuer_address,
         }
 
 

app/model/schema/company_info.py

@@ -26,6 +26,7 @@
 from app.model.schema.token_coupon import RetrieveCouponTokenResponse
 from app.model.schema.token_membership import RetrieveMembershipTokenResponse
 from app.model.schema.token_share import RetrieveShareTokenResponse
+from app.model.type.company_list import Trustee
 
 
 ############################
@@ -34,6 +35,7 @@
 class CompanyInfo(BaseModel):
     address: EthereumAddress
     corporate_name: str
+    trustee: Trustee | None = None
     rsa_publickey: str
     homepage: str
 

app/model/schema/public_info.py

@@ -36,6 +36,7 @@
 class TokenBase(BaseModel):
     token_address: EthereumAddress
     key_manager: list[str]
+    issuer_address: EthereumAddress | None
 
 
 class IbetBondToken(TokenBase):

app/model/type/__init__.py

@@ -0,0 +1,22 @@
+# pyright: reportUnusedImport=false
+"""
+Copyright BOOSTRY Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+from .company_list import CompanyListItem
+from .token_list import TokenListItem

app/model/type/company_list.py

@@ -0,0 +1,52 @@
+# cSpell:ignore publickey BOOSTRY
+"""
+Copyright BOOSTRY Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+from typing import Any
+
+from eth_utils.address import to_checksum_address
+from pydantic import BaseModel, Field, field_validator
+
+from app.model.schema.base import EthereumAddress
+
+
+class Trustee(BaseModel):
+    corporate_name: str = Field(..., min_length=1, max_length=30)
+    corporate_number: str
+    corporate_address: str = Field(..., min_length=1, max_length=60)
+
+
+class CompanyListItem(BaseModel):
+    address: EthereumAddress
+    corporate_name: str
+    trustee: Trustee | None = None
+    rsa_publickey: str
+    homepage: str = ""
+
+    @field_validator("homepage", mode="before")
+    @classmethod
+    def fill_missing_homepage(cls, value: Any) -> str:
+        if value is None:
+            return ""
+        return value
+
+    @field_validator("address")
+    @classmethod
+    def convert_to_checksum(cls, value: EthereumAddress) -> EthereumAddress:
+        return to_checksum_address(value)

app/model/type/token_list.py

@@ -0,0 +1,42 @@
+"""
+Copyright BOOSTRY Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+from typing import Literal
+
+from eth_utils.address import to_checksum_address
+from pydantic import BaseModel, field_validator
+
+from app.model.schema.base import EthereumAddress
+
+
+class TokenListItem(BaseModel):
+    token_template: Literal["ibetBond", "ibetShare", "ibetMembership", "ibetCoupon"]
+    product_type: int
+    token_address: EthereumAddress
+    key_manager: list[str]
+    issuer_address: EthereumAddress | None = None
+
+    @field_validator("token_address", "issuer_address")
+    @classmethod
+    def convert_to_checksum(
+        cls, value: EthereumAddress | None
+    ) -> EthereumAddress | None:
+        if value is None:
+            return value
+        return to_checksum_address(value)

app/utils/company_list.py

@@ -30,7 +30,13 @@
 
 
 class CompanyList:
-    DEFAULT = {"address": "", "corporate_name": "", "rsa_publickey": "", "homepage": ""}
+    DEFAULT = {
+        "address": "",
+        "corporate_name": "",
+        "rsa_publickey": "",
+        "homepage": "",
+        "trustee": None,
+    }
 
     @classmethod
     async def get(cls):
@@ -123,12 +129,17 @@ def rsa_publickey(self):
     def homepage(self):
         return self.obj.get("homepage")
 
+    @property
+    def trustee(self):
+        return self.obj.get("trustee")
+
     def json(self):
         return {
             "address": self.obj.get("address") or "",
             "corporate_name": self.obj.get("corporate_name") or "",
             "rsa_publickey": self.obj.get("rsa_publickey") or "",
             "homepage": self.obj.get("homepage") or "",
+            "trustee": self.obj.get("trustee"),
         }
 
     def __getitem__(self, key: str):

batch/indexer_Company_List.py

@@ -23,7 +23,7 @@
 import time
 
 import requests
-from eth_utils import to_checksum_address
+from pydantic import ValidationError
 from requests.adapters import HTTPAdapter
 from sqlalchemy import delete
 from sqlalchemy.engine.create import create_engine
@@ -39,6 +39,7 @@
 )
 from app.errors import ServiceUnavailable
 from app.model.db import Company
+from app.model.type import CompanyListItem
 from batch import free_malloc, log
 
 process_name = "INDEXER-COMPANY-LIST"
@@ -57,6 +58,9 @@ def process(self):
         LOG.info("Syncing company list")
 
         # Get from COMPANY_LIST_URL
+        if COMPANY_LIST_URL is None:
+            LOG.warning("COMPANY_LIST_URL is not set")
+            return
         try:
             with requests.Session() as session:
                 adapter = HTTPAdapter(max_retries=Retry(3, allowed_methods=["GET"]))
@@ -91,34 +95,20 @@ def process(self):
 
             # Insert company list
             for i, company in enumerate(company_list_json):
-                address = company.get("address", "")
-                corporate_name = company.get("corporate_name", "")
-                rsa_publickey = company.get("rsa_publickey", "")
-                homepage = (
-                    company.get("homepage")
-                    if company.get("homepage") is not None
-                    else ""
-                )
+                try:
+                    company_list_item = CompanyListItem.model_validate(company)  # type: ignore[arg-type]
+                except (ValidationError, ValueError):
+                    LOG.notice(f"Invalid company data: index={i} company={company}")
+                    continue
+
                 if (
-                    not isinstance(address, str)
-                    or not isinstance(corporate_name, str)
-                    or not isinstance(rsa_publickey, str)
-                    or not isinstance(homepage, str)
+                    company_list_item.address
+                    and company_list_item.corporate_name
+                    and company_list_item.rsa_publickey
                 ):
-                    LOG.notice(f"Invalid type: index={i}")
-                    continue
-                if address and corporate_name and rsa_publickey:
-                    try:
-                        address = to_checksum_address(address)
-                    except ValueError:
-                        LOG.notice(f"Invalid address: index={i} address={address}")
-                        continue
                     self.__sink_on_company(
                         db_session=db_session,
-                        address=to_checksum_address(address),
-                        corporate_name=corporate_name,
-                        rsa_publickey=rsa_publickey,
-                        homepage=homepage,
+                        company_list_item=company_list_item,
                     )
                 else:
                     LOG.notice(f"Missing required field: index={i}")
@@ -135,16 +125,21 @@ def process(self):
     @staticmethod
     def __sink_on_company(
         db_session: Session,
-        address: str,
-        corporate_name: str,
-        rsa_publickey: str,
-        homepage: str,
+        company_list_item: CompanyListItem,
     ):
         _company = Company()
-        _company.address = address
-        _company.corporate_name = corporate_name
-        _company.rsa_publickey = rsa_publickey
-        _company.homepage = homepage
+        _company.address = company_list_item.address
+        _company.corporate_name = company_list_item.corporate_name
+        _company.rsa_publickey = company_list_item.rsa_publickey
+        _company.homepage = company_list_item.homepage
+        if company_list_item.trustee:
+            _company.trustee_corporate_name = company_list_item.trustee.corporate_name
+            _company.trustee_corporate_number = (
+                company_list_item.trustee.corporate_number
+            )
+            _company.trustee_corporate_address = (
+                company_list_item.trustee.corporate_address
+            )
         db_session.merge(_company)