plan (1212f49a) · Commits · csst-dfs / csst-dfs-client

csst_dfs_client/common/init.py

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -59,7 +59,7 @@ class Result(dict):
	self["code"] = resp.status_code		self["code"] = resp.status_code
	if self.success:		if self.success:
	content_dict = json.loads(resp.content.decode('utf-8'))		content_dict = json.loads(resp.content.decode('utf-8'))
	if 'total_count' in content_dict and 'records' in content_dict:		if isinstance(content_dict, dict) and 'total_count' in content_dict and 'records' in content_dict:
	self["data"] = content_dict['records']		self["data"] = content_dict['records']
	self["total_count"] = content_dict['total_count']		self["total_count"] = content_dict['total_count']
	else:		else:

csst_dfs_client/level1.py

+11 −1

Original line number	Original line	Diff line number	Diff line
	from typing import Optional, IO, Tuple, Literal		from typing import Optional, Tuple, Literal
	from .common import request, Result		from .common import request, Result
	import os		import os

	@@ -99,6 +99,16 @@ def find_by_level1_id(level1_id: str) -> Result:
	return request.get(f"/api/level1/{level1_id}")		return request.get(f"/api/level1/{level1_id}")

	def find_by_brick_id(brick_id: int) -> Result:		def find_by_brick_id(brick_id: int) -> Result:
			"""
			通过 brick 的 ID 查询1级数据。

			Args:
			brick_id (int): 天区ID。

			Returns:
			Result: 查询结果。

			"""
	return request.get(f"/api/level1/brick/{brick_id}")		return request.get(f"/api/level1/brick/{brick_id}")

	def sls_find_by_qc1_status(qc1_status: int, limit: int = 1) -> Result:		def sls_find_by_qc1_status(qc1_status: int, limit: int = 1) -> Result:

csst_dfs_client/level2.py

+1 −1

Original line number	Original line	Diff line number	Diff line
	import os		import os
	from typing import Optional, IO, Tuple, Literal		from typing import Optional, Tuple, Literal
	from .common import request, Result		from .common import request, Result

	DateTimeTuple = Tuple[str, str]		DateTimeTuple = Tuple[str, str]

csst_dfs_client/plan.py

0 → 100644

+94 −0

Original line number	Original line	Diff line number	Diff line
			import os
			from typing import Optional, IO, Tuple, Literal
			from .common import request, Result

			DateTimeTuple = Tuple[str, str]

			def find(mode: Optional[str] = None,
			obsid: Optional[str] = None,
			backend: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'] = 'MSC',
			obstype: Optional[str] = None,
			obstime: Optional[DateTimeTuple] = None,
			page: int = 1,
			limit: int = 0) -> Result:
			"""
			根据给定的参数在DFS中搜索编排数据。

			Args:
			mode (Optional[str], optional): 观测模式. Defaults to None.
			obsid (Optional[str], optional): 观测ID，支持模糊搜索. Defaults to None.
			backend (Optional[str], optional): 模块ID，如'MSC', 'IFS'. Defaults to None.
			obstype (Optional[str], optional): 观测类型，如'SCIE'. Defaults to None.
			obstime (Optional[DateTimeTuple], optional): 观测时间范围. Defaults to None.
			page (int, optional): 页码. Defaults to 1.
			limit (int, optional): 每页数量. Defaults to 0，不限制.

			Returns:
			Result: 搜索结果对象.

			"""

			params = {
			'mode': mode,
			'obsid': obsid,
			'backend': backend,
			'obstype': obstype,
			'obs_time_start': None,
			'obs_time_end': None,
			'create_time_start': None,
			'create_time_end': None,
			'page': page,
			'limit': limit,
			}

			if obstime is not None:
			params['obs_time_start'], params['obs_time_end'] = obstime

			return request.post("/api/plan", params)

			def get_by_id(_id: str) -> Result:
			return request.get(f"/api/plan/_id/{_id}")

			def find_by_opid(opid: str) -> Result:
			"""
			通过 plan 的 opid 查询编排数据。

			Args:
			opid (str): opid。

			Returns:
			Result: 查询结果。

			"""
			return request.get(f"/api/plan/{opid}")

			def write_file(local_file: Optional[IO \| str], **kwargs) -> Result:
			"""
			将本地json文件或json数据流写入DFS中。

			Args:
			local_file (str]): 文件路径。
			**kwargs: 额外的关键字参数，这些参数将传递给DFS。

			Returns:
			Result: 操作的结果对象，包含操作是否成功以及相关的错误信息，成功返回数据对象。

			"""
			if isinstance(local_file, str):
			if not os.path.exists(local_file):
			raise FileNotFoundError(local_file)
			return request.post_file("/api/plan/file", local_file, kwargs)
			return request.post_bytesio("/api/plan/file", local_file, kwargs)

			def new(data: dict) -> Result:
			"""
			新建编排数据

			Args:
			data (dict): 编排数据的字典表示，如：{'opid': 'xxx', 'backend': 'MSC', ...}

			Returns:
			Result: 成功后，Result.data为写入记录，失败message为失败原因。

			"""
			return request.post("/api/plan/new", data)
			No newline at end of file

requirements-test.txt

+1 −0

Original line number	Original line	Diff line number	Diff line
	requests==2.31.0		requests==2.31.0
	bumpver==2023.1129		bumpver==2023.1129
			boto3==1.35.24
	# for testing		# for testing
	pytest==7.4.3		pytest==7.4.3
	coverage==7.6.1		coverage==7.6.1