Add ListAllIssuedTokens API

app/main.py

@@ -45,6 +45,7 @@
     position,
     settlement_issuer,
     share,
+    token_common,
     token_holders,
 )
 from app.routers.misc import (
@@ -148,6 +149,7 @@ async def root():
     app.include_router(notification.router)
     app.include_router(position.router)
     app.include_router(share.router)
+    app.include_router(token_common.router)
     app.include_router(token_holders.router)
     app.include_router(settlement_issuer.router)
     app.include_router(sealed_tx.router)

app/model/schema/__init__.py

@@ -171,6 +171,8 @@
     ListAllAdditionalIssueUploadQuery,
     ListAllHoldersQuery,
     ListAllHoldersSortItem,
+    ListAllIssuedTokensQuery,
+    ListAllIssuedTokensResponse,
     ListAllRedeemUploadQuery,
     ListAllTokenLockEventsQuery,
     ListAllTokenLockEventsResponse,

app/model/schema/token.py

@@ -39,6 +39,7 @@
     MMDD_constr,
     ResultSet,
     SortOrder,
+    TokenType,
     ValueOperator,
     YYYYMMDD_constr,
 )
@@ -308,6 +309,27 @@ class IbetShareRedeem(BaseModel):
     amount: int = Field(..., ge=1, le=1_000_000_000_000)
 
 
+class ListAllIssuedTokensSortItem(StrEnum):
+    CREATED = "created"
+    TOKEN_ADDRESS = "token_address"
+
+
+class ListAllIssuedTokensQuery(BasePaginationQuery):
+    """ListAllIssuedTokens query parameters"""
+
+    token_address_list: Optional[list[EthereumAddress]] = Field(
+        None, description="Token address to filter (**this affects total number**)"
+    )
+    token_type: Optional[TokenType] = Field(None, description="Token type")
+
+    sort_item: Optional[ListAllIssuedTokensSortItem] = Field(
+        ListAllIssuedTokensSortItem.CREATED
+    )
+    sort_order: Optional[SortOrder] = Field(
+        SortOrder.DESC, description=SortOrder.__doc__
+    )
+
+
 class IssueRedeemSortItem(StrEnum):
     """Issue/Redeem sort item"""
 
@@ -461,6 +483,27 @@ class ListTokenOperationLogHistoryQuery(BasePaginationQuery):
 ############################
 # RESPONSE
 ############################
+class IssuedToken(BaseModel):
+    """Issued Token"""
+
+    issuer_address: str = Field(description="Issuer address")
+    token_address: str = Field(description="Token address")
+    token_type: TokenType = Field(description="Token type")
+    created: str = Field(description="Created(Issued) datetime")
+    token_status: Optional[int] = Field(description="Token status")
+    contract_version: str = Field(description="Contract version")
+    token_attributes: IbetStraightBond | IbetShare = Field(
+        description="Token attributes"
+    )
+
+
+class ListAllIssuedTokensResponse(BaseModel):
+    """List issued tokens schema"""
+
+    result_set: ResultSet
+    tokens: list[IssuedToken]
+
+
 class TokenAddressResponse(BaseModel):
     """token address"""
 

app/routers/issuer/token_common.py

@@ -0,0 +1,135 @@
+"""
+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 Annotated, Optional
+
+import pytz
+from fastapi import APIRouter, Header, Query
+from sqlalchemy import asc, desc, func, select
+
+import config
+from app.database import DBAsyncSession
+from app.model.blockchain import IbetShareContract, IbetStraightBondContract
+from app.model.db import Token, TokenType
+from app.model.schema import ListAllIssuedTokensQuery, ListAllIssuedTokensResponse
+from app.utils.check_utils import address_is_valid_address, validate_headers
+from app.utils.docs_utils import get_routers_responses
+from app.utils.fastapi_utils import json_response
+
+router = APIRouter(
+    prefix="",
+    tags=["token_common"],
+)
+
+local_tz = pytz.timezone(config.TZ)
+utc_tz = pytz.timezone("UTC")
+
+
+# GET: /tokens
+@router.get(
+    "/tokens",
+    operation_id="ListAllIssuedTokens",
+    response_model=ListAllIssuedTokensResponse,
+    responses=get_routers_responses(422),
+)
+async def list_all_issued_tokens(
+    db: DBAsyncSession,
+    request_query: Annotated[ListAllIssuedTokensQuery, Query()],
+    issuer_address: Annotated[Optional[str], Header()] = None,
+):
+    """List all tokens issued from ibet-Prime"""
+
+    # Validate Headers
+    validate_headers(issuer_address=(issuer_address, address_is_valid_address))
+
+    # Base Query
+    if issuer_address is None:
+        stmt = select(Token)
+    else:
+        stmt = select(Token).where(Token.issuer_address == issuer_address)
+
+    if request_query.token_address_list is not None:
+        stmt = stmt.where(Token.token_address.in_(request_query.token_address_list))
+
+    total = await db.scalar(select(func.count()).select_from(stmt.subquery()))
+
+    # Search Filter
+    if request_query.token_type is not None:
+        stmt = stmt.where(Token.type == request_query.token_type)
+
+    count = await db.scalar(select(func.count()).select_from(stmt.subquery()))
+
+    # Sort
+    sort_attr = getattr(Token, request_query.sort_item, None)
+    if request_query.sort_order == 0:  # ASC
+        stmt = stmt.order_by(asc(sort_attr))
+    else:  # DESC
+        stmt = stmt.order_by(desc(sort_attr))
+
+    if request_query.sort_item != "created":
+        # NOTE: Set secondary sort for consistent results
+        stmt = stmt.order_by(desc(Token.created))
+
+    # Pagination
+    if request_query.limit is not None:
+        stmt = stmt.limit(request_query.limit)
+    if request_query.offset is not None:
+        stmt = stmt.offset(request_query.offset)
+
+    # Execute Query
+    issued_tokens = (await db.scalars(stmt)).all()
+
+    # Get Token Attributes
+    tokens = []
+    for _token in issued_tokens:
+        token_attr = None
+        if _token.type == TokenType.IBET_STRAIGHT_BOND:
+            token_attr = await IbetStraightBondContract(_token.token_address).get()
+        elif _token.type == TokenType.IBET_SHARE:
+            token_attr = await IbetShareContract(_token.token_address).get()
+
+        _issue_datetime = (
+            pytz.timezone("UTC")
+            .localize(_token.created)
+            .astimezone(local_tz)
+            .isoformat()
+        )
+
+        tokens.append(
+            {
+                "issuer_address": _token.issuer_address,
+                "token_address": _token.token_address,
+                "token_type": _token.type,
+                "created": _issue_datetime,
+                "token_status": _token.token_status,
+                "contract_version": _token.version,
+                "token_attributes": token_attr.__dict__,
+            }
+        )
+
+    resp = {
+        "result_set": {
+            "count": count,
+            "offset": request_query.offset,
+            "limit": request_query.limit,
+            "total": total,
+        },
+        "tokens": tokens,
+    }
+    return json_response(resp)