init

api/fund.py

-import uuid
+from typing import Union
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 from motor.core import AgnosticCollection
+from pymongo import ReturnDocument
 from starlette.requests import Request
 
 from exception.db import NotFundError
-from model import success_res, BaseJsonResponse
-from model.fund import CreateFund, response_fund_model, ResFund
+from model import success_res
+from model.fund import ApiCreateFund, DBCreateStakingFund, response_staking_fund_model, response_fund_model, FundType, \
+    DBCreateFund, ApiUpdateFund
+from tools.jwt_tools import get_current_user
 
 router = APIRouter()
 
 
-@router.post('/', response_model=BaseJsonResponse, summary='创建基金', description='创建基金')
-async def create(fund: CreateFund, request: Request):
-    collection: AgnosticCollection = request.app.state.mongodb_manger.get_client(name='pyfund', db='pyfund',
-                                                                                 collect='fund')
-    await collection.insert_one(fund.dict())
-    return success_res(data=fund.dict())
+def get_mongodb_client(request, db='pyfund', collect='fund') -> AgnosticCollection:
+    collection: AgnosticCollection = request.app.state.mongodb_manger.get_client(name='pyfund', db=db, collect=collect)
+    return collection
 
 
-@router.get('/{id}/', response_model=response_fund_model, summary='查询基金', description='查询基金')
-async def get(id: str, request: Request):
-    collection: AgnosticCollection = request.app.state.mongodb_manger.get_client(name='pyfund', db='pyfund',
-                                                                                 collect='fund')
-    data = await collection.find_one({'id': uuid.UUID(id)})
+@router.post('/', response_model=Union[response_staking_fund_model, response_fund_model], summary='创建基金',
+             description='创建基金')
+async def create(fund: ApiCreateFund, request: Request):
+    if fund.fund_type == FundType.staking:
+        # add_fund = DBCreateStakingFund(**fund.dict(), **user_payload)
+        add_fund = DBCreateStakingFund(**fund.dict())
+    else:
+        # add_fund = DBCreateFund.from_orm(**fund.dict(), **user_payload)
+        add_fund = DBCreateFund(**fund.dict())
+    collection = get_mongodb_client(request)
+    await collection.insert_one(add_fund.dict())
+    return success_res(data=add_fund)
+
+
+@router.put('/{fund_id}/', response_model=Union[response_fund_model, response_staking_fund_model], summary='更新基金',
+            description='更新基金')
+async def update(fund_id: str, update_fund: ApiUpdateFund, request: Request):
+    collection = get_mongodb_client(request)
+    data = await collection.find_one_and_update({'id': fund_id}, {'$set': update_fund.dict(exclude_unset=True)},
+                                                return_document=ReturnDocument.AFTER)
+    return success_res(data=data)
+
+
+@router.get('/{fund_id}/', response_model=Union[response_staking_fund_model], summary='查询基金',
+            description='查询基金')
+async def get(fund_id: str, request: Request):
+    collection = get_mongodb_client(request)
+    # data = await collection.find_one({'id': fund_id, 'user_id': user_payload['user_id']})
+    data = await collection.find_one({'id': fund_id})
     if not data:
         raise NotFundError()
-    return success_res(data=ResFund(**data))
+    if data['fund_type'] == FundType.staking:
+        res = DBCreateStakingFund(**data)
+    else:
+        res = DBCreateFund(**data)
+    return success_res(data=res)

configs/LOCAL.py

+from configs import env
+
+
 class Settings:
+    env = env
     name = '本地环境'
+    mongodb = 'mongodb://127.0.0.1:27017'
+    # 'mongodb://13.115.26.128:27017' # test
+
+    # api
+    identify_jwt = 'https://apit.matrixone.io/identity/.well-known/openid-configuration/jwks'

exception/db.py

@@ -6,3 +6,5 @@ from exception import MyException
 
 class NotFundError(MyException):
     status_code = status.HTTP_404_NOT_FOUND
+    default_error = '未找到数据'
+

main.py

+import json
 import traceback
 
+import httpx as httpx
 import uvicorn as uvicorn
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
+from jwt.algorithms import get_default_algorithms
 from loguru import logger
 from starlette import status
 from starlette.requests import Request
 
+import configs
 from api import api_router
 from configs import settings
 from db.mongodb_helper import AioMongodbManager
 from exception import MyException
 from model import error_response
+from tools.jwt_tools import get_identify_key
 
 app = FastAPI()
 
 mongodb_manger = AioMongodbManager()
-mongodb_manger.setup_pool('mongodb://13.115.26.128:27017', 'pyfund')
+mongodb_manger.setup_pool(settings.mongodb, 'pyfund')
 app.state.mongodb_manger = mongodb_manger
 # 添加路由
 app.include_router(api_router)
@@ -46,5 +51,10 @@ async def sys_exception_handler(request: Request, exc: Exception):
                           status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
 
+@app.on_event('startup')
+async def startup():
+    await get_identify_key()
+
+
 if __name__ == '__main__':
     uvicorn.run('main:app', host='0.0.0.0', port=8000)

model/__init__.py

-from typing import Any, List, Optional
+import datetime
+import json
+import uuid
+from typing import Any, List, Optional, TypeVar, Generic
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, validator
+from pydantic.generics import GenericModel
 from starlette import status
 from starlette.responses import JSONResponse
 
+DataT = TypeVar('DataT')
+
+
+class DataModel(BaseModel):
+    numbers: list[int]
+    people: list[str]
+
+
+class Response(GenericModel, Generic[DataT]):
+    data: DataT | None
+    message: str = 'success'
+    status: int
+
 
 class BaseJsonResponse(BaseModel):
     data: Any
@@ -67,3 +84,22 @@ def error_response(message='failed', status_code=status.HTTP_500_INTERNAL_SERVER
         "status": status_code
     }
     return JSONResponse(content=res, status_code=status_code)
+
+
+class BaseCreateModel(BaseModel):
+    id: str = Field(default_factory=uuid.uuid1().__str__, description='唯一ID')
+    create_time: int = Field(default_factory=lambda: int(datetime.datetime.utcnow().timestamp()),
+                             description='创建时间')
+    update_time: int = Field(default_factory=lambda: int(datetime.datetime.utcnow().timestamp()),
+                             description='更新时间')
+
+    class Config:
+        orm_mode = True
+
+
+class BaseUpdateModel(BaseModel):
+    update_time: int = Field(default_factory=lambda: int(datetime.datetime.utcnow().timestamp()),
+                             description='更新时间')
+
+    class Config:
+        orm_mode = True

model/fund.py

-import datetime
-import uuid
-from typing import List
+from enum import Enum
+from typing import List, Optional
 
 from pydantic import BaseModel, Field
 
-from model import dynamic_response
+from model import dynamic_response, BaseCreateModel, BaseUpdateModel, Response
+from model.node import DBNode
 
 
-class Node(BaseModel):
-    pub_key: str
-    create_time: datetime.datetime
+class FundType(str, Enum):
+    staking = 'staking'
+    other = 'other'
 
 
-class CreateFund(BaseModel):
-    id: uuid.UUID = Field(default_factory=uuid.uuid1, description='基金ID')
+class ApiCreateFund(BaseModel):
     name: str = Field(..., description='基金名称')
-    settlement_time: str = Field(default='08:00')
-    nav: float = Field(default=1)
-    nodes: List[Node] = Field(default=[])
-    create_time: datetime.datetime = Field(default_factory=datetime.datetime.utcnow)
+    fund_type: FundType = Field(default=FundType.staking, description='基金类型')
+    base_coin: str = Field(default='USD', description='基准币种')
+    base_nav: float = Field(default=1, description='初始净值')
+    settlement_time: str = Field(default='08:00', description='结算时间')
+    # nodes: List[DBNode] = Field(default=[], description='绑定节点')
 
+    class Config:
+        orm_mode = True
 
-class ResFund(CreateFund):
+
+class DBCreateFund(ApiCreateFund, BaseCreateModel):
+    user_id: str = Field(None, description='创建人')
+    user_email: str = Field(None, description='创建人')
+    nav: float = Field(default=1, description='当前净值')
+
+
+class DBCreateStakingFund(DBCreateFund):
+    nodes: List[DBNode] = Field(default=[], description='绑定节点')
+
+
+class ApiUpdateFund(BaseModel):
+    name: Optional[str] = Field(None, description='基金名称')
+    fund_type: Optional[FundType] = Field(default=None, description='基金类型')
+    base_coin: Optional[str] = Field(None, description='基准币种')
+    base_nav: Optional[float] = Field(None, description='初始净值')
+    settlement_time: Optional[str] = Field(None, description='结算时间')
+
+    class Config:
+        orm_mode = True
+
+
+class DBUpdateFund(ApiUpdateFund, BaseUpdateModel):
     pass
 
 
-response_fund_model = dynamic_response(ResFund)
+# response_fund_model = dynamic_response(DBCreateFund)
+response_fund_model = Response[DBCreateFund]
+response_staking_fund_model = Response[DBCreateStakingFund]
+# response_staking_fund_model = dynamic_response(DBCreateStakingFund)
 # response_list_branch_model = list_dynamic_response(ResBranch)

model/node.py

+from enum import Enum
+
+from pydantic import Field, BaseModel
+
+from model import BaseCreateModel
+
+
+class NodeStatus(str, Enum):
+    pending = 'pending'
+    active = 'active'
+    stop = 'stop'
+
+
+class ApiCreateNode(BaseModel):
+    pub_key: str = Field(..., description='创建人')
+    status: NodeStatus = Field(default=NodeStatus.pending, description='创建人')
+
+
+class DBNode(BaseCreateModel, ApiCreateNode):
+    pass
+
+# response_node_model = dynamic_response(DBCreateFund)
+# response_list_branch_model = list_dynamic_response(ResBranch)

requirements.txt

@@ -3,8 +3,11 @@ fastapi==0.92.0
 loguru==0.6.0
 aiohttp==3.8.4
 aioredis==2.0.1
-motor==2.1
+motor==3.1.1
 pytz==2022.7.1
 PyJWT==2.6.0
 pydantic==1.10.6
-starlette==0.25.0
+starlette==0.25.0
\ No newline at end of file
+cryptography=39.0.2
+pymongo~=4.3.3
+httpx~=0.23.3
\ No newline at end of file

tools/jwt_tools.py

 import json
+from pathlib import Path
+from urllib.parse import urlparse
+
+import httpx
 import jwt
 from fastapi import Security
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
@@ -9,35 +13,38 @@ from loguru import logger
 
 from configs import settings
 from exception.token import RequestPubKeyError, TokenError
-from tools.http_helper import aio_request
 
 security = HTTPBearer()
 
 
-async def req_pub_key(url):
+async def get_identify_key():
     """
-    请求公钥
+     生成公钥
     :return:
     """
     # 请求key
+    content = ''
     try:
-        res = await aio_request(url=url)
-        res = res['result']
-        logger.info('获取key成功 {}'.format(res))
-        raw_key_data = res['keys'][0]
-        algorithms = raw_key_data['alg']
-        default_pub_key = get_default_algorithms()[algorithms]
-        public_key = default_pub_key.from_jwk(json.dumps(raw_key_data))
-        return algorithms, public_key
+        async with httpx.AsyncClient() as client:
+            response = await client.get(settings.identify_jwt)
+            content = response.content
+            res = json.loads(content)
+            print('公钥获取成功')
+            key_data = res['keys'][0]
+            rsa = get_default_algorithms()[key_data['alg']]
+            public_key = rsa.from_jwk(json.dumps(key_data))
+            settings.public_key = public_key
+            settings.algorithms = key_data['alg']
     except Exception as e:
-        raise RequestPubKeyError(f'jwt key获取失败, url:{url}  {e}')
+        raise RequestPubKeyError(f'公钥获取失败, url:{settings.identify_jwt}, res:{content} {e}')
 
 
 def decode_token(token):
     payload = jwt.decode(token,
-                         key=settings.secret_key,
+                         key=settings.public_key,
+                         issuer=f'http://{urlparse(settings.identify_jwt).netloc}',
                          algorithms=[settings.algorithms],
-                         options={'verify_iss': False, 'verify_aud': False})
+                         options={'verify_iss': False if settings.env == 'LOCAL' else True, 'verify_aud': False})
     return payload
 
 
@@ -46,8 +53,9 @@ def get_current_user(credentials: HTTPAuthorizationCredentials = Security(securi
     try:
         assert credentials.scheme == 'Bearer'
         payload = decode_token(token)  # options={'verify_signature':False}
-        user_id: str = payload.get("id")
-        if user_id is None:
+        payload['user_id'] = payload.pop("id", None)
+        payload['user_email'] = payload.pop("email", None)
+        if payload['user_id'] is None:
             raise TokenError('错误的Token')
         return payload
     except ExpiredSignatureError: