12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455 |
- from typing import Dict, List, Optional, Tuple
- import arrow
- import numpy as np
- from httpx import AsyncClient, URL
- from loguru import logger
- from app.core.config import settings
- from app.services.service import Service
- from app.utils.date import get_time_str, TIME_FMT
- class RealWorldService(Service):
- def __init__(
- self,
- client: AsyncClient,
- group_code: str,
- project_id: str,
- app_id: Optional[str] = None,
- user_id: Optional[str] = None,
- server_settings=settings
- ):
- super(RealWorldService, self).__init__(client)
- self._group_code = group_code
- self._project_id = project_id
- self._app_id = app_id
- self._user_id = user_id
- self._base_url = URL(server_settings.PLATFORM_HOST)
- self._now_time = get_time_str()
- def _common_parameters(self) -> Dict:
- return {
- 'groupCode': self._group_code,
- 'projectId': self._project_id
- }
- async def query_instance(
- self,
- object_id: Optional[List] = None,
- class_code: Optional[str] = None
- ):
- url = self._base_url.join('/rwd/instance/object/query')
- params = self._common_parameters()
- criteria = dict()
- if object_id:
- criteria.update({'id': object_id})
- if class_code:
- criteria.update({'classCode': class_code})
- payload = {
- 'criteria': criteria
- }
- raw_info = await self._post(url, params, payload)
- return raw_info
|