rwd.py 1.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. from typing import Dict, List, Optional, Tuple
  2. import arrow
  3. import numpy as np
  4. from httpx import AsyncClient, URL
  5. from loguru import logger
  6. from app.core.config import settings
  7. from app.services.service import Service
  8. from app.utils.date import get_time_str, TIME_FMT
  9. class RealWorldService(Service):
  10. def __init__(
  11. self,
  12. client: AsyncClient,
  13. group_code: str,
  14. project_id: str,
  15. app_id: Optional[str] = None,
  16. user_id: Optional[str] = None,
  17. server_settings=settings
  18. ):
  19. super(RealWorldService, self).__init__(client)
  20. self._group_code = group_code
  21. self._project_id = project_id
  22. self._app_id = app_id
  23. self._user_id = user_id
  24. self._base_url = URL(server_settings.PLATFORM_HOST)
  25. self._now_time = get_time_str()
  26. def _common_parameters(self) -> Dict:
  27. return {
  28. 'groupCode': self._group_code,
  29. 'projectId': self._project_id
  30. }
  31. async def query_instance(
  32. self,
  33. object_id: Optional[List] = None,
  34. class_code: Optional[str] = None
  35. ):
  36. url = self._base_url.join('/rwd/instance/object/query')
  37. params = self._common_parameters()
  38. criteria = dict()
  39. if object_id:
  40. criteria.update({'id': object_id})
  41. if class_code:
  42. criteria.update({'classCode': class_code})
  43. payload = {
  44. 'criteria': criteria
  45. }
  46. raw_info = await self._post(url, params, payload)
  47. return raw_info