1234567891011121314151617181920212223242526272829303132333435363738394041424344454647 |
- 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
|