feat: sap-cli skill v2.1.0 — self-contained distributable package
- assets/: sap-cli source (v2.1.0, 26 commands, 16 object types) - references/: tool constraints + error handling (self-contained) - scripts/setup.py: one-click install/config/verify - SKILL.md: full command reference + dual-platform install guide - VERSION: 2.1.0 Built from D:/Codespace/sap-cli via scripts/pack_skill.py
This commit is contained in:
@@ -0,0 +1,29 @@
|
||||
"""sapcli.client — ADT client split into Mixin modules.
|
||||
|
||||
Usage::
|
||||
|
||||
from sapcli.client import ADTClient
|
||||
|
||||
The public API is identical to the original monolithic ``client.py``.
|
||||
"""
|
||||
|
||||
from sapcli.client._base import ADTClientBase
|
||||
from sapcli.client._source import SourceMixin
|
||||
from sapcli.client._transport import TransportMixin
|
||||
from sapcli.client._search import SearchMixin
|
||||
from sapcli.client._ddic import DdicMixin
|
||||
|
||||
|
||||
class ADTClient(
|
||||
ADTClientBase,
|
||||
SourceMixin,
|
||||
TransportMixin,
|
||||
SearchMixin,
|
||||
DdicMixin,
|
||||
):
|
||||
"""Full ADT client composed from domain-specific Mixins."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
__all__ = ["ADTClient"]
|
||||
@@ -0,0 +1,88 @@
|
||||
"""ADTClient base — connection, authentication, and generic request helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import requests
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
||||
from sapcli.exceptions import LoginError
|
||||
|
||||
logger = logging.getLogger("sapcli.client")
|
||||
|
||||
|
||||
class ADTClientBase:
|
||||
"""Connection / authentication foundation shared by all mixins."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
host: str,
|
||||
client: str,
|
||||
user: str,
|
||||
password: str,
|
||||
verify_ssl: bool = False,
|
||||
) -> None:
|
||||
self.host: str = host.rstrip("/")
|
||||
self.csrf_token: str = "fetch"
|
||||
self.session: requests.Session = requests.Session()
|
||||
self.session.auth = HTTPBasicAuth(user, password)
|
||||
self.session.verify = verify_ssl
|
||||
self._stateful: bool = False
|
||||
self.sap_client: str = client
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Context-manager protocol
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def __enter__(self): # type: ignore[override]
|
||||
self.login()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
|
||||
self.session.close()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Headers / auth helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _headers(self, content_type: str = "application/xml") -> dict[str, str]:
|
||||
return {
|
||||
"Accept": "*/*",
|
||||
"Cache-Control": "no-cache",
|
||||
"x-csrf-token": self.csrf_token,
|
||||
"X-sap-adt-sessiontype": "stateful" if self._stateful else "stateless",
|
||||
"content-type": content_type,
|
||||
"sap-client": self.sap_client,
|
||||
}
|
||||
|
||||
def login(self) -> bool:
|
||||
url = f"{self.host}/sap/bc/adt/compatibility/graph"
|
||||
hdrs = self._headers()
|
||||
logger.info("LOGIN: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("LOGIN RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code == 200:
|
||||
self.csrf_token = resp.headers.get("x-csrf-token", "")
|
||||
if self.csrf_token:
|
||||
logger.info(
|
||||
"CSRF Token: %s...(%d chars)",
|
||||
self.csrf_token[:8],
|
||||
len(self.csrf_token),
|
||||
)
|
||||
return True
|
||||
raise LoginError(f"登录失败: HTTP {resp.status_code}")
|
||||
|
||||
def object_exists(self, obj_uri: str) -> bool:
|
||||
url = f"{self.host}{obj_uri}"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "*/*"
|
||||
logger.info("CHECK EXISTS: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("CHECK EXISTS RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code == 200:
|
||||
return True
|
||||
if resp.status_code == 404:
|
||||
return False
|
||||
logger.warning("Unexpected status %d for existence check", resp.status_code)
|
||||
return resp.status_code < 400
|
||||
@@ -0,0 +1,756 @@
|
||||
"""DdicMixin — object CRUD, DDIC helpers, CDS, packages, ATC, pretty-print."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import xml.etree.ElementTree as ET
|
||||
from typing import Any
|
||||
|
||||
from sapcli.exceptions import CreateError, DeleteError
|
||||
from sapcli.types import ObjectTypeConfig, get_type_config
|
||||
|
||||
logger = logging.getLogger("sapcli.client")
|
||||
|
||||
|
||||
class DdicMixin:
|
||||
"""Object creation / deletion, DDIC operations, CDS, packages, ATC, pretty-print."""
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Object create / delete
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def delete_object(
|
||||
self,
|
||||
obj_uri: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> tuple[bool, str]:
|
||||
lock_handle, _ = self.lock(obj_uri, corr_nr)
|
||||
url = f"{self.host}{obj_uri}"
|
||||
params: dict[str, str] = {"lockHandle": lock_handle}
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
hdrs = self._headers()
|
||||
logger.info("DELETE: DELETE %s", url)
|
||||
resp = self.session.delete(url, headers=hdrs, params=params)
|
||||
logger.info("DELETE RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code in (200, 204):
|
||||
logger.info("删除成功")
|
||||
return True, ""
|
||||
error_text = resp.text[:500] if resp.text else f"HTTP {resp.status_code}"
|
||||
logger.error("删除失败: %s", error_text)
|
||||
raise DeleteError(f"删除失败: {error_text}")
|
||||
|
||||
def create_object(
|
||||
self,
|
||||
obj_type: str,
|
||||
name: str,
|
||||
description: str | None = None,
|
||||
corr_nr: str | None = None,
|
||||
source: str | None = None,
|
||||
package: str = "$TMP",
|
||||
) -> tuple[str, str | None]:
|
||||
config = get_type_config(obj_type)
|
||||
params: dict[str, str] = {}
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
|
||||
if obj_type == "function":
|
||||
if "/" not in name:
|
||||
raise CreateError("function 类型需要'函数组名/函数模块名' 格式")
|
||||
group, fm = name.split("/", 1)
|
||||
collection_url = config.format_collection_uri(group=group.lower())
|
||||
obj_name = fm.upper()
|
||||
obj_uri = config.format_obj_uri(fm.lower(), group=group.lower())
|
||||
src_uri = config.format_src_uri(fm.lower(), group=group.lower())
|
||||
else:
|
||||
collection_url = config.collection_uri
|
||||
obj_name = name.upper()
|
||||
obj_uri = config.format_obj_uri(name.lower())
|
||||
src_uri = config.format_src_uri(name.lower())
|
||||
|
||||
body = self._build_create_body(obj_type, obj_name, description, package)
|
||||
url = f"{self.host}{collection_url}"
|
||||
hdrs = self._headers(config.create_content_type)
|
||||
logger.info("CREATE: POST %s name=%s", url, obj_name)
|
||||
resp = self.session.post(
|
||||
url, headers=hdrs, params=params, data=body.encode("utf-8")
|
||||
)
|
||||
logger.info("CREATE RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("CREATE ERROR: %s", resp.text[:2000])
|
||||
raise CreateError(
|
||||
f"创建对象失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
|
||||
logger.info("对象已创建: %s", obj_uri)
|
||||
|
||||
if source and src_uri:
|
||||
lock_handle, _ = self.lock(obj_uri, corr_nr)
|
||||
try:
|
||||
self.set_source(src_uri, source, lock_handle, corr_nr)
|
||||
finally:
|
||||
self.unlock(obj_uri, lock_handle)
|
||||
self.activate(obj_name, obj_uri, corr_nr)
|
||||
|
||||
return obj_uri, src_uri
|
||||
|
||||
def _build_create_body(
|
||||
self,
|
||||
obj_type: str,
|
||||
name: str,
|
||||
description: str | None,
|
||||
package: str = "$TMP",
|
||||
) -> str:
|
||||
desc = description or name
|
||||
if obj_type == "report":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<program:abapProgram xmlns:program="http://www.sap.com/adt/programs/programs"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="PROG/P"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</program:abapProgram>"
|
||||
)
|
||||
elif obj_type == "class":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<class:abapClass xmlns:class="http://www.sap.com/adt/oo/classes"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="CLAS/OC"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</class:abapClass>"
|
||||
)
|
||||
elif obj_type == "function":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<fm:abapFunctionModule xmlns:fm="http://www.sap.com/adt/functions/fmodules"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="FUNC/F"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</fm:abapFunctionModule>"
|
||||
)
|
||||
elif obj_type == "functiongroup":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<fg:abapFunctionGroup xmlns:fg="http://www.sap.com/adt/functions/groups"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="FUGR/F"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</fg:abapFunctionGroup>"
|
||||
)
|
||||
elif obj_type == "interface":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<intf:abapInterface xmlns:intf="http://www.sap.com/adt/oo/interfaces"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="INTF/OI"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</intf:abapInterface>"
|
||||
)
|
||||
elif obj_type == "domain":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<doma:domain xmlns:doma="http://www.sap.com/dictionary/domain"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="DOMA/DD"'
|
||||
f' adtcore:description="{desc}"'
|
||||
' adtcore:language="EN">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</doma:domain>"
|
||||
)
|
||||
elif obj_type == "dataelement":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<dtel:dataElement xmlns:dtel="http://www.sap.com/adt/dictionary/dataelements"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="DTEL/DE"'
|
||||
f' adtcore:description="{desc}"'
|
||||
' adtcore:language="EN">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</dtel:dataElement>"
|
||||
)
|
||||
elif obj_type == "table":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<tabl:table xmlns:tabl="http://www.sap.com/adt/dictionary/tables"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="TABL/TT"'
|
||||
f' adtcore:description="{desc}"'
|
||||
' adtcore:language="EN">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</tabl:table>"
|
||||
)
|
||||
elif obj_type == "tabletype":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<ttyp:abapTableType xmlns:ttyp="http://www.sap.com/adt/ddic/tabletypes"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="TTYP"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</ttyp:abapTableType>"
|
||||
)
|
||||
raise ValueError(f"不支持的对象类型: {obj_type}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# DDIC helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def create_ddic_object(
|
||||
self,
|
||||
obj_type: str,
|
||||
name: str,
|
||||
definition_body: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> tuple[str, str | None]:
|
||||
config = get_type_config(obj_type)
|
||||
obj_uri = config.format_obj_uri(name.lower())
|
||||
src_uri = config.format_src_uri(name.lower())
|
||||
|
||||
if obj_type in ("table", "structure"):
|
||||
self._put_ddl_source(src_uri, definition_body, "", corr_nr)
|
||||
else:
|
||||
obj_name = name.upper()
|
||||
body = self._build_create_body(obj_type, obj_name, obj_name)
|
||||
collection_url = config.collection_uri
|
||||
params: dict[str, str] = {}
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
hdrs = self._headers(config.create_content_type)
|
||||
logger.info(
|
||||
"CREATE DDIC: POST %s name=%s (with full definition)",
|
||||
collection_url,
|
||||
obj_name,
|
||||
)
|
||||
resp = self.session.post(
|
||||
f"{self.host}{collection_url}",
|
||||
headers=hdrs,
|
||||
params=params,
|
||||
data=definition_body.encode("utf-8"),
|
||||
)
|
||||
logger.info("CREATE DDIC RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
raise CreateError(
|
||||
f"创建 DDIC 对象失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
|
||||
obj_name = name.upper()
|
||||
success, messages = self.activate(obj_name, obj_uri, corr_nr)
|
||||
if not success:
|
||||
errors = [m for m in messages if m["type"] == "E"]
|
||||
if errors:
|
||||
raise CreateError(
|
||||
f"DDIC 对象激活失败: {'; '.join(e['text'] for e in errors)}"
|
||||
)
|
||||
|
||||
logger.info("DDIC 对象已创建并激活: %s", obj_uri)
|
||||
return obj_uri, src_uri
|
||||
|
||||
def _put_ddic_xml(
|
||||
self,
|
||||
obj_uri: str,
|
||||
xml_body: str,
|
||||
lock_handle: str = "",
|
||||
corr_nr: str | None = None,
|
||||
) -> bool:
|
||||
url = f"{self.host}{obj_uri}"
|
||||
params: dict[str, str] = {}
|
||||
if lock_handle:
|
||||
params["lockHandle"] = lock_handle
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "*/*"
|
||||
logger.info("PUT DDIC XML: PUT %s (%d chars)", url, len(xml_body))
|
||||
resp = self.session.put(
|
||||
url, headers=hdrs, params=params, data=xml_body.encode("utf-8")
|
||||
)
|
||||
logger.info("PUT DDIC XML RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
raise CreateError(
|
||||
f"写入 DDIC XML 失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
return True
|
||||
|
||||
def _put_ddl_with_auto_lock(
|
||||
self,
|
||||
src_uri: str,
|
||||
ddl_body: str,
|
||||
obj_uri: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> bool:
|
||||
try:
|
||||
lock_handle, _ = self.lock(obj_uri, corr_nr)
|
||||
except Exception:
|
||||
logger.debug("自动锁定失败，使用空锁句柄继续", exc_info=True)
|
||||
lock_handle = ""
|
||||
try:
|
||||
return self.set_source(src_uri, ddl_body, lock_handle, corr_nr)
|
||||
finally:
|
||||
if lock_handle:
|
||||
self.unlock(obj_uri, lock_handle)
|
||||
|
||||
def _put_ddl_source(
|
||||
self,
|
||||
src_uri: str,
|
||||
ddl_body: str,
|
||||
lock_handle: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> bool:
|
||||
return self.set_source(src_uri, ddl_body, lock_handle, corr_nr)
|
||||
|
||||
def get_object_status(self, obj_uri: str) -> dict[str, Any]:
|
||||
"""查询对象在 SAP 系统中的状态。
|
||||
|
||||
Returns:
|
||||
{"exists": bool, "status": str, "corr_nr": str|None}
|
||||
status: "active" / "inactive" / "not_exists"
|
||||
"""
|
||||
url = f"{self.host}{obj_uri}"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "*/*"
|
||||
logger.info("GET OBJECT STATUS: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET OBJECT STATUS RESPONSE: HTTP %s", resp.status_code)
|
||||
|
||||
if resp.status_code == 404:
|
||||
return {"exists": False, "status": "not_exists", "corr_nr": None}
|
||||
|
||||
if resp.status_code != 200:
|
||||
logger.warning(
|
||||
"Unexpected status %d for object status check: %s",
|
||||
resp.status_code,
|
||||
obj_uri,
|
||||
)
|
||||
return {"exists": False, "status": "not_exists", "corr_nr": None}
|
||||
|
||||
# 解析 XML 获取 version（active/inactive）
|
||||
status = "active"
|
||||
corr_nr: str | None = None
|
||||
try:
|
||||
root = ET.fromstring(resp.content)
|
||||
ns = {"adtcore": "http://www.sap.com/adt/core"}
|
||||
version = root.attrib.get(f"{{{ns['adtcore']}}}version", "")
|
||||
if version == "inactive":
|
||||
status = "inactive"
|
||||
elif version and version != "active":
|
||||
status = version
|
||||
except ET.ParseError:
|
||||
pass
|
||||
|
||||
# 尝试获取 corr_nr：通过快速 lock → unlock 探测
|
||||
try:
|
||||
lock_handle, detected_corr = self.lock(obj_uri)
|
||||
corr_nr = detected_corr
|
||||
self.unlock(obj_uri, lock_handle)
|
||||
except Exception:
|
||||
# 锁定失败也正常（可能权限问题），corr_nr 保持 None
|
||||
logger.debug("探测 corr_nr 失败", exc_info=True)
|
||||
|
||||
return {"exists": True, "status": status, "corr_nr": corr_nr}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Function-group helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def function_group_exists(self, group_name: str) -> bool:
|
||||
url = f"{self.host}/sap/bc/adt/functions/groups/{group_name.lower()}"
|
||||
hdrs = self._headers()
|
||||
logger.info("CHECK FG EXISTS: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("CHECK FG EXISTS RESPONSE: HTTP %s", resp.status_code)
|
||||
return resp.status_code == 200
|
||||
|
||||
def create_function_group(
|
||||
self,
|
||||
group_name: str,
|
||||
description: str | None = None,
|
||||
corr_nr: str | None = None,
|
||||
) -> bool:
|
||||
config = get_type_config("functiongroup")
|
||||
params: dict[str, str] = {"groupname": group_name.upper()}
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
desc = description or group_name
|
||||
body = self._build_create_body("functiongroup", group_name.upper(), desc)
|
||||
url = f"{self.host}{config.collection_uri}"
|
||||
hdrs = self._headers(config.create_content_type)
|
||||
logger.info("CREATE FG: POST %s name=%s", url, group_name.upper())
|
||||
resp = self.session.post(
|
||||
url, headers=hdrs, params=params, data=body.encode("utf-8")
|
||||
)
|
||||
logger.info("CREATE FG RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("CREATE FG ERROR: %s", resp.text[:2000])
|
||||
raise CreateError(
|
||||
f"创建函数组失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
return True
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# CDS View
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_cds_source(self, name: str) -> str:
|
||||
"""读取 CDS View DDL 源码。
|
||||
|
||||
Args:
|
||||
name: CDS 名称。
|
||||
|
||||
Returns:
|
||||
DDL 源码字符串。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/ddic/ddlsources/{name.lower()}/source/main"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "text/plain"
|
||||
logger.info("GET CDS SOURCE: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET CDS SOURCE RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
return resp.text
|
||||
|
||||
def create_cds(
|
||||
self,
|
||||
name: str,
|
||||
description: str,
|
||||
ddl_source: str,
|
||||
) -> tuple[str, str]:
|
||||
"""创建 CDS View 并写入 DDL 源码。
|
||||
|
||||
Args:
|
||||
name: CDS 名称。
|
||||
description: 描述。
|
||||
ddl_source: DDL 源码。
|
||||
|
||||
Returns:
|
||||
``(obj_uri, src_uri)`` 元组。
|
||||
"""
|
||||
obj_uri = f"/sap/bc/adt/ddic/ddlsources/{name.lower()}"
|
||||
src_uri = f"/sap/bc/adt/ddic/ddlsources/{name.lower()}/source/main"
|
||||
|
||||
# 1) 创建 CDS 对象
|
||||
create_url = f"{self.host}/sap/bc/adt/ddic/ddlsources"
|
||||
desc = description or name
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<ddlsrc:abapSource xmlns:ddlsrc="http://www.sap.com/adt/ddic/ddlsources"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name.lower()}"'
|
||||
' adtcore:type="DDLS/DF"'
|
||||
f' adtcore:description="{desc}">'
|
||||
'<adtcore:packageRef adtcore:name="$TMP"/>'
|
||||
"</ddlsrc:abapSource>"
|
||||
)
|
||||
hdrs = self._headers("application/xml")
|
||||
params: dict[str, str] = {"name": name.lower()}
|
||||
logger.info("CREATE CDS: POST %s name=%s", create_url, name)
|
||||
resp = self.session.post(
|
||||
create_url,
|
||||
headers=hdrs,
|
||||
params=params,
|
||||
data=body.encode("utf-8"),
|
||||
)
|
||||
logger.info("CREATE CDS RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("CREATE CDS ERROR: %s", resp.text[:2000])
|
||||
raise CreateError(
|
||||
f"创建 CDS 失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
|
||||
# 2) 写入 DDL 源码
|
||||
lock_handle, corr_nr = self.lock(obj_uri)
|
||||
try:
|
||||
self.set_source(src_uri, ddl_source, lock_handle, corr_nr)
|
||||
finally:
|
||||
self.unlock(obj_uri, lock_handle)
|
||||
|
||||
return obj_uri, src_uri
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Package
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def create_package(
|
||||
self,
|
||||
name: str,
|
||||
description: str | None = None,
|
||||
superpackage: str | None = None,
|
||||
) -> bool:
|
||||
"""创建 ABAP 包。
|
||||
|
||||
Args:
|
||||
name: 包名。
|
||||
description: 描述。
|
||||
superpackage: 上级包名。
|
||||
|
||||
Returns:
|
||||
是否成功。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/packages"
|
||||
desc = description or name
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<pkg:package xmlns:pkg="http://www.sap.com/adt/packages"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:description="{desc}"'
|
||||
)
|
||||
if superpackage:
|
||||
body += f'><adtcore:packageRef adtcore:name="{superpackage}"/></pkg:package>'
|
||||
else:
|
||||
body += "/>"
|
||||
|
||||
hdrs = self._headers("application/xml")
|
||||
logger.info("CREATE PACKAGE: POST %s name=%s", url, name)
|
||||
resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
|
||||
logger.info("CREATE PACKAGE RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("CREATE PACKAGE ERROR: %s", resp.text[:2000])
|
||||
raise CreateError(
|
||||
f"创建包失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
return True
|
||||
|
||||
def get_package_info(self, name: str) -> dict[str, str]:
|
||||
"""获取 ABAP 包信息。
|
||||
|
||||
Args:
|
||||
name: 包名。
|
||||
|
||||
Returns:
|
||||
字典含 ``name``, ``description``, ``owner``, ``superpackage`` 等。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/packages/{name}"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "application/xml"
|
||||
logger.info("GET PACKAGE INFO: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET PACKAGE INFO RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
core_ns = "http://www.sap.com/adt/core"
|
||||
info: dict[str, str] = {
|
||||
"name": name,
|
||||
"description": "",
|
||||
"owner": "",
|
||||
"superpackage": "",
|
||||
}
|
||||
# 尝试从根元素属性提取
|
||||
info["description"] = root.attrib.get(f"{{{core_ns}}}description", "")
|
||||
info["owner"] = root.attrib.get(f"{{{core_ns}}}owner", "")
|
||||
# 上级包引用
|
||||
pkg_ref = root.find(f".//{{{core_ns}}}packageRef")
|
||||
if pkg_ref is not None:
|
||||
info["superpackage"] = pkg_ref.attrib.get(f"{{{core_ns}}}name", "")
|
||||
return info
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# ATC / Quality
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def atc_check(
|
||||
self,
|
||||
name: str,
|
||||
obj_uri: str,
|
||||
variant: str | None = None,
|
||||
) -> tuple[bool, list[dict[str, str]]]:
|
||||
"""执行 ATC 代码检查。
|
||||
|
||||
Args:
|
||||
name: 对象名称。
|
||||
obj_uri: 对象 ADT URI。
|
||||
variant: 检查变体名称。
|
||||
|
||||
Returns:
|
||||
``(success, findings)`` — success 表示无严重错误，
|
||||
findings 是发现项列表，每项含 ``type``, ``line``, ``text`` 等。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/atos/checks"
|
||||
params: dict[str, str] = {"context": obj_uri}
|
||||
if variant:
|
||||
params["variant"] = variant
|
||||
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
|
||||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
|
||||
"</adtcore:objectReferences>"
|
||||
)
|
||||
|
||||
hdrs = self._headers()
|
||||
logger.info("ATC CHECK: POST %s name=%s", url, name)
|
||||
resp = self.session.post(
|
||||
url, headers=hdrs, params=params, data=body.encode("utf-8")
|
||||
)
|
||||
logger.info("ATC CHECK RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
findings: list[dict[str, str]] = []
|
||||
has_error = False
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
core_ns = "http://www.sap.com/adt/core"
|
||||
for item in root.iter():
|
||||
severity = item.attrib.get("severity", item.attrib.get("type", ""))
|
||||
line = item.attrib.get("line", "")
|
||||
text = item.attrib.get("message", item.text or "")
|
||||
if severity or text:
|
||||
findings.append({"type": severity, "line": line, "text": text})
|
||||
if severity in ("E", "1"):
|
||||
has_error = True
|
||||
|
||||
return not has_error, findings
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Pretty printer
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def pretty_print(self, source: str) -> str:
|
||||
"""调用 ABAP Pretty Printer 格式化源码。
|
||||
|
||||
Args:
|
||||
source: 原始 ABAP 源码。
|
||||
|
||||
Returns:
|
||||
格式化后的源码。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/prettyprinter"
|
||||
hdrs = self._headers("text/plain; charset=utf-8")
|
||||
hdrs["Accept"] = "text/plain"
|
||||
logger.info("PRETTY PRINT: POST %s (%d chars)", url, len(source))
|
||||
resp = self.session.post(url, headers=hdrs, data=source.encode("utf-8"))
|
||||
logger.info("PRETTY PRINT RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
return resp.text
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Table structure / data query
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_table_fields(self, table_name: str) -> list[dict[str, str]]:
|
||||
"""查询 DDIC 表的字段结构。
|
||||
|
||||
Args:
|
||||
table_name: 表名（不区分大小写）。
|
||||
|
||||
Returns:
|
||||
字段列表，每项含 name, type, length, description, key_attribute。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/datapreview/ddic/{table_name.lower()}/metadata"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "*/*"
|
||||
logger.info("GET TABLE FIELDS: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET TABLE FIELDS RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
ns = {"dp": "http://www.sap.com/adt/dataPreview"}
|
||||
fields = []
|
||||
for col in root.findall(".//dp:columns/dp:metadata", ns):
|
||||
field = {
|
||||
"name": col.attrib.get(f"{{{ns['dp']}}}name", ""),
|
||||
"type": col.attrib.get(f"{{{ns['dp']}}}type", ""),
|
||||
"length": col.attrib.get(f"{{{ns['dp']}}}length", ""),
|
||||
"description": col.attrib.get(f"{{{ns['dp']}}}description", ""),
|
||||
"key_attribute": col.attrib.get(f"{{{ns['dp']}}}keyAttribute", "false"),
|
||||
}
|
||||
fields.append(field)
|
||||
return fields
|
||||
|
||||
def query_table_data(self, sql: str, max_rows: int = 200) -> dict:
|
||||
"""通过 ADT freestyle SQL 查询表数据。
|
||||
|
||||
Args:
|
||||
sql: SELECT SQL 语句。
|
||||
max_rows: 最大返回行数。
|
||||
|
||||
Returns:
|
||||
{"columns": [...字段名...], "rows": [[值1, 值2, ...], ...], "total_rows": int, "execution_time": str}
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/datapreview/freestyle"
|
||||
hdrs = self._headers("text/plain; charset=utf-8")
|
||||
hdrs["Accept"] = "*/*"
|
||||
params = {"rowNumber": str(max_rows)}
|
||||
logger.info("QUERY TABLE DATA: POST %s sql=%s maxRows=%d", url, sql[:80], max_rows)
|
||||
resp = self.session.post(url, headers=hdrs, params=params, data=sql.encode("utf-8"))
|
||||
logger.info("QUERY TABLE DATA RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
ns = {"dp": "http://www.sap.com/adt/dataPreview"}
|
||||
|
||||
# 提取列名（从第一组 columns/metadata）
|
||||
columns = []
|
||||
for col in root.findall(".//dp:columns/dp:metadata", ns):
|
||||
name = col.attrib.get(f"{{{ns['dp']}}}name", "")
|
||||
if name:
|
||||
columns.append(name)
|
||||
|
||||
# 提取数据 — ADT 按列存储（每个 columns 包含一列的 dataSet）
|
||||
# 需要转置为按行返回
|
||||
col_data: list[list[str]] = []
|
||||
for col_group in root.findall(".//dp:columns", ns):
|
||||
dataset = col_group.find("dp:dataSet", ns)
|
||||
if dataset is None:
|
||||
col_data.append([])
|
||||
continue
|
||||
values = [v.text or "" for v in dataset.findall("dp:data", ns)]
|
||||
col_data.append(values)
|
||||
|
||||
# 转置：列数据 → 行数据
|
||||
max_len = max((len(c) for c in col_data), default=0)
|
||||
rows = []
|
||||
for i in range(max_len):
|
||||
row = []
|
||||
for c in col_data:
|
||||
row.append(c[i] if i < len(c) else "")
|
||||
rows.append(row)
|
||||
|
||||
# 元数据
|
||||
total_el = root.find("dp:totalRows", ns)
|
||||
time_el = root.find("dp:queryExecutionTime", ns)
|
||||
|
||||
return {
|
||||
"columns": columns,
|
||||
"rows": rows,
|
||||
"total_rows": int(total_el.text) if total_el is not None and total_el.text else len(rows),
|
||||
"execution_time": time_el.text if time_el is not None else "",
|
||||
}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Run program
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run_program(self, program_name: str) -> str:
|
||||
"""远程执行 ABAP 程序并返回输出。
|
||||
|
||||
Args:
|
||||
program_name: 程序名（不区分大小写）。
|
||||
|
||||
Returns:
|
||||
程序的标准输出文本（text/plain）。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/programs/programrun/{program_name.lower()}"
|
||||
hdrs = self._headers("application/xml")
|
||||
hdrs["Accept"] = "*/*"
|
||||
logger.info("RUN PROGRAM: POST %s", url)
|
||||
resp = self.session.post(url, headers=hdrs)
|
||||
logger.info("RUN PROGRAM RESPONSE: HTTP %s (%d bytes)", resp.status_code, len(resp.content))
|
||||
resp.raise_for_status()
|
||||
return resp.text
|
||||
@@ -0,0 +1,197 @@
|
||||
"""SearchMixin — repository search, where-used, code search, diff helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from sapcli.exceptions import SapCliError
|
||||
|
||||
logger = logging.getLogger("sapcli.client")
|
||||
|
||||
|
||||
class SearchMixin:
|
||||
"""Object search, where-used listing, and source-code search."""
|
||||
|
||||
def list_objects(
|
||||
self,
|
||||
obj_type: str | None = None,
|
||||
package: str | None = None,
|
||||
prefix: str | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
"""通过 ADT Information System 搜索对象。
|
||||
|
||||
Args:
|
||||
obj_type: ADT 类型代码，例如 ``PROG/P``, ``CLAS/OC``。
|
||||
package: 包名过滤。
|
||||
prefix: 名称前缀过滤（如 ``Z*``）。
|
||||
|
||||
Returns:
|
||||
列表，每项含 ``name``, ``type``, ``description``, ``package``。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/repository/informationsystem/search"
|
||||
params: dict[str, str] = {"maxrow": "200"}
|
||||
if obj_type:
|
||||
params["type"] = obj_type
|
||||
if prefix:
|
||||
params["name"] = prefix
|
||||
else:
|
||||
params["name"] = "*"
|
||||
if package:
|
||||
params["pkg"] = package
|
||||
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "application/xml"
|
||||
logger.info("LIST OBJECTS: GET %s params=%s", url, params)
|
||||
resp = self.session.get(url, headers=hdrs, params=params)
|
||||
logger.info("LIST OBJECTS RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
results: list[dict[str, str]] = []
|
||||
core_ns = "http://www.sap.com/adt/core"
|
||||
for ref in root.findall(f".//{{{core_ns}}}objectReference"):
|
||||
name = ref.attrib.get(f"{{{core_ns}}}name", "")
|
||||
otype = ref.attrib.get(f"{{{core_ns}}}type", "")
|
||||
desc = ref.attrib.get(f"{{{core_ns}}}description", "")
|
||||
pkg = ref.attrib.get(f"{{{core_ns}}}package", "")
|
||||
results.append(
|
||||
{
|
||||
"name": name,
|
||||
"type": otype,
|
||||
"description": desc,
|
||||
"package": pkg,
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
def where_used(
|
||||
self,
|
||||
name: str,
|
||||
obj_uri: str,
|
||||
adt_type: str | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
"""Where-Used 引用查询。
|
||||
|
||||
Args:
|
||||
name: 对象名称。
|
||||
obj_uri: 对象 ADT URI。
|
||||
adt_type: ADT 类型代码。
|
||||
|
||||
Returns:
|
||||
列表，每项含 ``name``, ``type``, ``package``, ``uri``。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/usage/whereusedlist"
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
|
||||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"'
|
||||
)
|
||||
if adt_type:
|
||||
body += f' adtcore:type="{adt_type}"'
|
||||
body += "/></adtcore:objectReferences>"
|
||||
|
||||
hdrs = self._headers()
|
||||
logger.info("WHERE USED: POST %s name=%s", url, name)
|
||||
resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
|
||||
logger.info("WHERE USED RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
results: list[dict[str, str]] = []
|
||||
core_ns = "http://www.sap.com/adt/core"
|
||||
for ref in root.findall(f".//{{{core_ns}}}objectReference"):
|
||||
ref_name = ref.attrib.get(f"{{{core_ns}}}name", "")
|
||||
ref_type = ref.attrib.get(f"{{{core_ns}}}type", "")
|
||||
ref_pkg = ref.attrib.get(f"{{{core_ns}}}package", "")
|
||||
ref_uri = ref.attrib.get(f"{{{core_ns}}}uri", "")
|
||||
results.append(
|
||||
{
|
||||
"name": ref_name,
|
||||
"type": ref_type,
|
||||
"package": ref_pkg,
|
||||
"uri": ref_uri,
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
def search_code(
|
||||
self,
|
||||
query: str,
|
||||
obj_type: str | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
"""源代码搜索。
|
||||
|
||||
Args:
|
||||
query: 搜索关键词。
|
||||
obj_type: ADT 类型代码过滤。
|
||||
|
||||
Returns:
|
||||
列表，每项含 ``name``, ``type``, ``description``。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/repository/structuredsearch"
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<ssearch:search xmlns:ssearch="http://www.sap.com/adt/structuredsearch"'
|
||||
' ssearch:maxResults="200">'
|
||||
"<ssearch:queries>"
|
||||
"<ssearch:query>"
|
||||
f"<ssearch:selection>"
|
||||
f'<ssearch:attribute ssearch:attribute="name" ssearch:operator="contains"'
|
||||
f' ssearch:value="{query}"/>'
|
||||
)
|
||||
if obj_type:
|
||||
body += (
|
||||
f'<ssearch:attribute ssearch:attribute="type" ssearch:operator="equals"'
|
||||
f' ssearch:value="{obj_type}"/>'
|
||||
)
|
||||
body += (
|
||||
"</ssearch:selection>"
|
||||
"</ssearch:query>"
|
||||
"</ssearch:queries>"
|
||||
"</ssearch:search>"
|
||||
)
|
||||
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "application/xml"
|
||||
logger.info("SEARCH CODE: POST %s query=%s", url, query)
|
||||
resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
|
||||
logger.info("SEARCH CODE RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
results: list[dict[str, str]] = []
|
||||
core_ns = "http://www.sap.com/adt/core"
|
||||
for ref in root.findall(f".//{{{core_ns}}}objectReference"):
|
||||
obj_name = ref.attrib.get(f"{{{core_ns}}}name", "")
|
||||
obj_type_val = ref.attrib.get(f"{{{core_ns}}}type", "")
|
||||
desc = ref.attrib.get(f"{{{core_ns}}}description", "")
|
||||
results.append(
|
||||
{
|
||||
"name": obj_name,
|
||||
"type": obj_type_val,
|
||||
"description": desc,
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Diff helper
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def read_source_for_diff(self, name: str, obj_type: str) -> str:
|
||||
"""读取对象源码用于 diff 对比。
|
||||
|
||||
Args:
|
||||
name: 对象名称。
|
||||
obj_type: 对象类型键（如 ``report``, ``class`` 等）。
|
||||
|
||||
Returns:
|
||||
源代码字符串。
|
||||
"""
|
||||
from sapcli.types import parse_object_name
|
||||
|
||||
parsed = parse_object_name(name, obj_type)
|
||||
if parsed.src_uri is None:
|
||||
raise SapCliError(f"{obj_type} 对象没有源代码 URI")
|
||||
return self.get_source(parsed.src_uri)
|
||||
@@ -0,0 +1,404 @@
|
||||
"""SourceMixin — source code read/write, lock/unlock, activate, syntax-check."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from sapcli.exceptions import (
|
||||
ActivationError,
|
||||
LockError,
|
||||
SyntaxCheckError,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("sapcli.client")
|
||||
|
||||
|
||||
class SourceMixin:
|
||||
"""Source-code read/write and lock management."""
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Source read / write
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_source(self, src_uri: str) -> str:
|
||||
url = f"{self.host}{src_uri}"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "text/plain"
|
||||
logger.info("GET SOURCE: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info(
|
||||
"GET SOURCE RESPONSE: HTTP %s, %d bytes",
|
||||
resp.status_code,
|
||||
len(resp.content) if resp.content else 0,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.text
|
||||
|
||||
def set_source(
|
||||
self,
|
||||
src_uri: str,
|
||||
source: str,
|
||||
lock_handle: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> bool:
|
||||
params: dict[str, str] = {}
|
||||
if lock_handle:
|
||||
params["lockHandle"] = lock_handle
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
url = f"{self.host}{src_uri}"
|
||||
hdrs = self._headers("text/plain; charset=utf-8")
|
||||
logger.info("SET SOURCE: PUT %s (%d chars)", url, len(source))
|
||||
resp = self.session.put(
|
||||
url, headers=hdrs, params=params, data=source.encode("utf-8")
|
||||
)
|
||||
logger.info("SET SOURCE RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("SET SOURCE ERROR BODY: %s", resp.text[:2000])
|
||||
existing_nr = self._extract_locked_corrnr(resp.text)
|
||||
if existing_nr and existing_nr != corr_nr:
|
||||
logger.info("使用 corrNr=%s 重试 SET SOURCE", existing_nr)
|
||||
params["corrNr"] = existing_nr
|
||||
resp = self.session.put(
|
||||
url, headers=hdrs, params=params, data=source.encode("utf-8")
|
||||
)
|
||||
logger.info("SET SOURCE RETRY RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
resp.raise_for_status()
|
||||
return True
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Lock / unlock
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def lock(
|
||||
self,
|
||||
obj_uri: str,
|
||||
corr_nr: str | None = None,
|
||||
accept: str | None = None,
|
||||
) -> tuple[str, str | None]:
|
||||
"""锁定对象。
|
||||
|
||||
Returns:
|
||||
(lock_handle, effective_corr_nr) — 实际使用的传输请求号。
|
||||
如果对象已绑定在某请求中，effective_corrnr 为该请求号；
|
||||
否则为传入的 corr_nr（可能为 None）。
|
||||
"""
|
||||
self._stateful = True
|
||||
url = f"{self.host}{obj_uri}"
|
||||
params: dict[str, str] = {"_action": "LOCK", "accessMode": "MODIFY"}
|
||||
effective_corr_nr = corr_nr
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = accept or "*/*"
|
||||
logger.info("LOCK: POST %s (corrNr=%s)", url, corr_nr or "none")
|
||||
resp = self.session.post(url, headers=hdrs, params=params)
|
||||
logger.info("LOCK RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code == 500:
|
||||
existing_nr = self._extract_locked_corrnr(resp.text)
|
||||
if existing_nr:
|
||||
effective_corr_nr = existing_nr
|
||||
logger.info("对象已锁定在请求 %s 中，使用该请求重试", existing_nr)
|
||||
params["corrNr"] = existing_nr
|
||||
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
||||
logger.info("LOCK RETRY RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
raise LockError(f"锁定失败: HTTP {resp.status_code}", obj_uri=obj_uri)
|
||||
root = ET.fromstring(resp.content)
|
||||
handle = root.findtext(".//LOCK_HANDLE")
|
||||
if not handle:
|
||||
for el in root.iter():
|
||||
if "HANDLE" in el.tag.upper():
|
||||
handle = el.text
|
||||
break
|
||||
if not handle:
|
||||
raise LockError("锁定失败: 未获取到 lock handle", obj_uri=obj_uri)
|
||||
|
||||
# 从成功响应中提取 corrNr（对象可能已绑定传输请求）
|
||||
if not effective_corr_nr:
|
||||
extracted = self._extract_corrnr_from_lock_response(root)
|
||||
if extracted:
|
||||
effective_corr_nr = extracted
|
||||
logger.info("从 lock 响应中检测到 corrNr: %s", extracted)
|
||||
|
||||
logger.info(
|
||||
"Lock handle: %s (corrNr=%s)",
|
||||
handle[:20] if len(handle) > 20 else handle,
|
||||
effective_corr_nr or "none",
|
||||
)
|
||||
return handle, effective_corr_nr
|
||||
|
||||
def _extract_locked_corrnr(self, error_body: str) -> str | None:
|
||||
try:
|
||||
root = ET.fromstring(error_body)
|
||||
for entry in root.iter():
|
||||
if entry.attrib.get("key") == "corrNr":
|
||||
corrnr = entry.text
|
||||
if corrnr and corrnr != "*":
|
||||
return corrnr
|
||||
for el in root.iter():
|
||||
if el.tag.endswith("message") and el.text:
|
||||
m = re.search(r"request\s+(\w+)", el.text, re.IGNORECASE)
|
||||
if m:
|
||||
return m.group(1)
|
||||
except Exception:
|
||||
logger.debug("XML 解析提取 locked corrnr 失败", exc_info=True)
|
||||
return None
|
||||
|
||||
def _extract_corrnr_from_lock_response(self, root: ET.Element) -> str | None:
|
||||
"""从成功的 lock 响应 XML 中提取传输请求号。
|
||||
|
||||
常见格式:
|
||||
<lock:lock ...>
|
||||
<lock:LOCK_HANDLE>...</lock:LOCK_HANDLE>
|
||||
<lock:CORRN>DEVK901362</lock:CORRN>
|
||||
</lock:lock>
|
||||
也可能是:
|
||||
<adtcore:property adtcore:key="corrNr">DEVK901362</adtcore:property>
|
||||
"""
|
||||
# 策略 1: 直接找 CORRN 标签
|
||||
for tag_name in ("CORRN", "corrNr", "corr_nr"):
|
||||
text = root.findtext(f".//{tag_name}")
|
||||
if text and text.strip() and text.strip() != "*":
|
||||
return text.strip()
|
||||
|
||||
# 策略 2: 遍历所有元素，找标签或属性包含 corrNr 的
|
||||
for el in root.iter():
|
||||
# 检查属性
|
||||
for attr_key in el.attrib:
|
||||
if "corr" in attr_key.lower() and "nr" in attr_key.lower():
|
||||
val = el.attrib[attr_key]
|
||||
if val and val.strip() and val.strip() != "*":
|
||||
return val.strip()
|
||||
# 检查标签名包含 CORRN
|
||||
if el.text and "corr" in el.tag.lower() and el.text.strip() and el.text.strip() != "*":
|
||||
return el.text.strip()
|
||||
|
||||
# 策略 3: 找 key="corrNr" 的 property 元素
|
||||
for el in root.iter():
|
||||
if el.attrib.get("key") == "corrNr" and el.text:
|
||||
val = el.text.strip()
|
||||
if val and val != "*":
|
||||
return val
|
||||
|
||||
return None
|
||||
|
||||
def unlock(self, obj_uri: str, lock_handle: str) -> bool:
|
||||
url = f"{self.host}{obj_uri}"
|
||||
params = {"_action": "UNLOCK", "lockHandle": lock_handle}
|
||||
hdrs = self._headers("text/plain; charset=utf-8")
|
||||
logger.info("UNLOCK: POST %s", url)
|
||||
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
||||
logger.info("UNLOCK RESPONSE: HTTP %s", resp.status_code)
|
||||
self._stateful = False
|
||||
return resp.status_code == 200
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Activate / syntax check
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def activate(
|
||||
self,
|
||||
name: str,
|
||||
obj_uri: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> tuple[bool, list[dict[str, str]]]:
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
|
||||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
|
||||
"</adtcore:objectReferences>"
|
||||
)
|
||||
self._stateful = False
|
||||
url = f"{self.host}/sap/bc/adt/activation"
|
||||
params: dict[str, str] = {"method": "activate"}
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
hdrs = self._headers()
|
||||
logger.info("ACTIVATE: POST %s name=%s corrNr=%s", url, name, corr_nr or "none")
|
||||
resp = self.session.post(url, headers=hdrs, params=params, data=body)
|
||||
logger.info(
|
||||
"ACTIVATE RESPONSE: HTTP %s CT=%s",
|
||||
resp.status_code,
|
||||
resp.headers.get("content-type", "")[:80],
|
||||
)
|
||||
|
||||
if resp.status_code != 200:
|
||||
logger.error("Activation HTTP %s", resp.status_code)
|
||||
raise ActivationError(f"激活失败: HTTP {resp.status_code}")
|
||||
|
||||
ct = resp.headers.get("content-type", "")
|
||||
if "inactivectsobjects" in ct:
|
||||
root = ET.fromstring(resp.content)
|
||||
ioc_ns = "http://www.sap.com/abapxml/inactiveCtsObjects"
|
||||
inactive_objects: list[str] = []
|
||||
for entry in root.findall(f".//{{{ioc_ns}}}entry"):
|
||||
obj_elem = entry.find(f"{{{ioc_ns}}}object")
|
||||
if obj_elem is not None:
|
||||
ref = obj_elem.find(".//{http://www.sap.com/adt/core}ref")
|
||||
if ref is None:
|
||||
for child in obj_elem:
|
||||
if child.tag.endswith("}ref"):
|
||||
ref = child
|
||||
break
|
||||
if ref is not None:
|
||||
inactive_name = ref.attrib.get(
|
||||
"{http://www.sap.com/adt/core}name", ""
|
||||
)
|
||||
inactive_type = ref.attrib.get(
|
||||
"{http://www.sap.com/adt/core}type", ""
|
||||
)
|
||||
if inactive_name:
|
||||
inactive_objects.append(f"{inactive_name} ({inactive_type})")
|
||||
if inactive_objects:
|
||||
logger.warning("激活返回未激活对象: %s", ", ".join(inactive_objects))
|
||||
msg_text = f"Objects still inactive: {', '.join(inactive_objects)}"
|
||||
messages: list[dict[str, str]] = [
|
||||
{"type": "E", "line": "?", "text": msg_text, "href": ""}
|
||||
]
|
||||
logger.info("Activation result: FAILED (inactive objects)")
|
||||
return False, messages
|
||||
logger.info("Activation result: SUCCESS (inactiveObjects 响应但无对象列出)")
|
||||
return True, []
|
||||
|
||||
if not resp.content or not resp.text.strip():
|
||||
# NW 7.40 返回空响应但可能并未真正激活
|
||||
# 验证对象实际状态
|
||||
logger.info("Activation returned empty response, verifying object status...")
|
||||
try:
|
||||
verify_resp = self.session.get(
|
||||
f"{self.host}{obj_uri}",
|
||||
headers=self._headers(),
|
||||
)
|
||||
if verify_resp.status_code == 200:
|
||||
vroot = ET.fromstring(verify_resp.content)
|
||||
version = vroot.attrib.get("{http://www.sap.com/adt/core}version", "")
|
||||
if version == "inactive":
|
||||
# NW 7.40: first activation may only "stage" the change.
|
||||
# Retry activation once, then re-verify.
|
||||
logger.info("Object still inactive, retrying activation (NW 7.40 double-activate workaround)...")
|
||||
retry_resp = self.session.post(
|
||||
url, headers=hdrs, params=params, data=body,
|
||||
)
|
||||
logger.info(
|
||||
"RETRY ACTIVATE: HTTP %s len=%s",
|
||||
retry_resp.status_code, len(retry_resp.content),
|
||||
)
|
||||
if retry_resp.status_code == 200:
|
||||
# Re-verify after retry
|
||||
verify_resp2 = self.session.get(
|
||||
f"{self.host}{obj_uri}",
|
||||
headers=self._headers(),
|
||||
)
|
||||
if verify_resp2.status_code == 200:
|
||||
vroot2 = ET.fromstring(verify_resp2.content)
|
||||
version2 = vroot2.attrib.get(
|
||||
"{http://www.sap.com/adt/core}version", ""
|
||||
)
|
||||
if version2 == "active":
|
||||
logger.info("Retry activation succeeded!")
|
||||
return True, []
|
||||
logger.warning("Object still inactive after retry (NW 7.40)")
|
||||
messages = [
|
||||
{
|
||||
"type": "W",
|
||||
"line": "?",
|
||||
"text": "ADT 激活返回空响应，对象仍为 inactive（NW 7.40 已知限制）。请在 SAP GUI SE09 手动激活。",
|
||||
"href": "",
|
||||
}
|
||||
]
|
||||
return False, messages
|
||||
elif version == "active":
|
||||
logger.info("Verified: object is active after activation")
|
||||
except Exception as e:
|
||||
logger.warning("Could not verify activation status: %s", e)
|
||||
logger.info("Activation success (空响应)")
|
||||
return True, []
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
messages = []
|
||||
for msg in root.iter():
|
||||
msg_type = msg.attrib.get("type", "")
|
||||
if msg_type in ("E", "W", "I", "S"):
|
||||
line = msg.attrib.get("line", "?")
|
||||
href = msg.attrib.get("href", "")
|
||||
txt = ""
|
||||
for child in msg.iter():
|
||||
if child.text and child.tag.endswith("}txt"):
|
||||
txt = child.text
|
||||
break
|
||||
if not txt:
|
||||
for child in msg.iter():
|
||||
if child.text and len(child.text.strip()) > 3:
|
||||
txt = child.text.strip()
|
||||
break
|
||||
messages.append({"type": msg_type, "line": line, "text": txt, "href": href})
|
||||
logger.info("Activation msg [%s] line=%s: %s", msg_type, line, txt)
|
||||
|
||||
errors = [m for m in messages if m["type"] == "E"]
|
||||
success = len(errors) == 0
|
||||
logger.info(
|
||||
"Activation result: %s (%d errors)",
|
||||
"SUCCESS" if success else "FAILED",
|
||||
len(errors),
|
||||
)
|
||||
return success, messages
|
||||
|
||||
def syntax_check(
|
||||
self,
|
||||
name: str,
|
||||
obj_uri: str,
|
||||
) -> tuple[bool, list[dict[str, str]]]:
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
|
||||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
|
||||
"</adtcore:objectReferences>"
|
||||
)
|
||||
self._stateful = False
|
||||
url = f"{self.host}/sap/bc/adt/activation"
|
||||
params = {"method": "check"}
|
||||
hdrs = self._headers()
|
||||
logger.info("SYNTAX CHECK: POST %s name=%s", url, name)
|
||||
resp = self.session.post(url, headers=hdrs, params=params, data=body)
|
||||
logger.info("SYNTAX CHECK RESPONSE: HTTP %s", resp.status_code)
|
||||
|
||||
if resp.status_code != 200:
|
||||
logger.error("Syntax check HTTP %s", resp.status_code)
|
||||
raise SyntaxCheckError(f"语法检查失败: HTTP {resp.status_code}")
|
||||
|
||||
if not resp.content or not resp.text.strip():
|
||||
logger.info("Syntax check: OK (空响应)")
|
||||
return True, []
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
messages: list[dict[str, str]] = []
|
||||
for msg in root.iter():
|
||||
msg_type = msg.attrib.get("type", "")
|
||||
if msg_type in ("E", "W", "I", "S"):
|
||||
line = msg.attrib.get("line", "?")
|
||||
href = msg.attrib.get("href", "")
|
||||
txt = ""
|
||||
for child in msg.iter():
|
||||
if child.text and child.tag.endswith("}txt"):
|
||||
txt = child.text
|
||||
break
|
||||
if not txt:
|
||||
for child in msg.iter():
|
||||
if child.text and len(child.text.strip()) > 3:
|
||||
txt = child.text.strip()
|
||||
break
|
||||
messages.append({"type": msg_type, "line": line, "text": txt, "href": href})
|
||||
logger.info("Syntax check msg [%s] line=%s: %s", msg_type, line, txt)
|
||||
|
||||
errors = [m for m in messages if m["type"] == "E"]
|
||||
success = len(errors) == 0
|
||||
logger.info(
|
||||
"Syntax check result: %s (%d errors)",
|
||||
"OK" if success else "ERRORS",
|
||||
len(errors),
|
||||
)
|
||||
return success, messages
|
||||
@@ -0,0 +1,203 @@
|
||||
"""TransportMixin — transport request management."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from sapcli.exceptions import CreateError, SapCliError
|
||||
|
||||
logger = logging.getLogger("sapcli.client")
|
||||
|
||||
|
||||
class TransportMixin:
|
||||
"""Transport-request CRUD and release."""
|
||||
|
||||
def get_transport_request(self) -> str | None:
|
||||
url = f"{self.host}/sap/bc/adt/cts/transportrequests"
|
||||
hdrs: dict[str, str] = {
|
||||
**self._headers(),
|
||||
"Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
|
||||
}
|
||||
logger.info("GET TRANSPORT: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET TRANSPORT RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
root = ET.fromstring(resp.content)
|
||||
tm_ns = "http://www.sap.com/cts/adt/tm"
|
||||
for req in root.findall(f".//{{{tm_ns}}}request"):
|
||||
num = req.attrib.get(f"{{{tm_ns}}}number", "")
|
||||
status = req.attrib.get(f"{{{tm_ns}}}status", "")
|
||||
if status == "D" and num:
|
||||
logger.info("Transport request: %s", num)
|
||||
return num
|
||||
return None
|
||||
|
||||
def list_transport_requests(self) -> list[dict[str, str]]:
|
||||
"""列出所有可修改的传输请求。
|
||||
|
||||
Returns:
|
||||
列表，每项包含 number, description, owner, status 字段。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/cts/transportrequests"
|
||||
hdrs: dict[str, str] = {
|
||||
**self._headers(),
|
||||
"Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
|
||||
}
|
||||
logger.info("LIST TRANSPORT: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("LIST TRANSPORT RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
root = ET.fromstring(resp.content)
|
||||
tm_ns = "http://www.sap.com/cts/adt/tm"
|
||||
results: list[dict[str, str]] = []
|
||||
for req in root.findall(f".//{{{tm_ns}}}request"):
|
||||
num = req.attrib.get(f"{{{tm_ns}}}number", "")
|
||||
status = req.attrib.get(f"{{{tm_ns}}}status", "")
|
||||
desc = req.attrib.get(f"{{{tm_ns}}}description", "")
|
||||
owner = req.attrib.get(f"{{{tm_ns}}}owner", "")
|
||||
if status == "D" and num:
|
||||
results.append(
|
||||
{
|
||||
"number": num,
|
||||
"description": desc,
|
||||
"owner": owner,
|
||||
"status": status,
|
||||
}
|
||||
)
|
||||
logger.info(
|
||||
"Transport request: %s — %s (owner: %s)", num, desc, owner
|
||||
)
|
||||
return results
|
||||
|
||||
def create_transport_request(self, description: str) -> str:
|
||||
"""创建新的传输请求。
|
||||
|
||||
Args:
|
||||
description: 传输请求描述。
|
||||
|
||||
Returns:
|
||||
新创建的传输请求编号。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/cts/transportrequests"
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="utf-8"?>'
|
||||
'<cts:request xmlns:cts="http://www.sap.com/adt/cts"'
|
||||
f' cts:type="K"'
|
||||
f' cts:description="{description}"/>'
|
||||
)
|
||||
hdrs = self._headers()
|
||||
logger.info("CREATE TRANSPORT: POST %s desc=%s", url, description)
|
||||
resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
|
||||
logger.info("CREATE TRANSPORT RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("CREATE TRANSPORT ERROR: %s", resp.text[:2000])
|
||||
raise CreateError(
|
||||
f"创建传输请求失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
|
||||
# 从响应中解析新请求编号
|
||||
root = ET.fromstring(resp.content)
|
||||
tm_ns = "http://www.sap.com/cts/adt/tm"
|
||||
for req in root.findall(f".//{{{tm_ns}}}request"):
|
||||
num = req.attrib.get(f"{{{tm_ns}}}number", "")
|
||||
if num:
|
||||
logger.info("New transport request created: %s", num)
|
||||
return num
|
||||
|
||||
# 尝试从 Location header 或响应体文本中提取编号
|
||||
location = resp.headers.get("Location", "")
|
||||
m = re.search(r"transportrequests/(\w+)", location)
|
||||
if m:
|
||||
logger.info("New transport request from Location: %s", m.group(1))
|
||||
return m.group(1)
|
||||
|
||||
logger.warning("无法从响应中解析新传输请求编号")
|
||||
return ""
|
||||
|
||||
def transport_info(self, corr_nr: str) -> dict[str, str]:
|
||||
"""获取传输请求详情。
|
||||
|
||||
Args:
|
||||
corr_nr: 传输请求编号。
|
||||
|
||||
Returns:
|
||||
字典含 ``number``, ``description``, ``status``, ``owner`` 等。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/cts/transportrequests/{corr_nr}"
|
||||
hdrs: dict[str, str] = {
|
||||
**self._headers(),
|
||||
"Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
|
||||
}
|
||||
logger.info("TRANSPORT INFO: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("TRANSPORT INFO RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
tm_ns = "http://www.sap.com/cts/adt/tm"
|
||||
info: dict[str, str] = {
|
||||
"number": corr_nr,
|
||||
"description": "",
|
||||
"status": "",
|
||||
"owner": "",
|
||||
}
|
||||
for req in root.findall(f".//{{{tm_ns}}}request"):
|
||||
info["number"] = req.attrib.get(f"{{{tm_ns}}}number", corr_nr)
|
||||
info["description"] = req.attrib.get(f"{{{tm_ns}}}description", "")
|
||||
info["status"] = req.attrib.get(f"{{{tm_ns}}}status", "")
|
||||
info["owner"] = req.attrib.get(f"{{{tm_ns}}}owner", "")
|
||||
break
|
||||
return info
|
||||
|
||||
def transport_release(self, corr_nr: str) -> bool:
|
||||
"""释放传输请求。
|
||||
|
||||
Args:
|
||||
corr_nr: 传输请求编号。
|
||||
|
||||
Returns:
|
||||
是否成功。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/cts/transportrequests/{corr_nr}/%20"
|
||||
hdrs = self._headers()
|
||||
params: dict[str, str] = {"_action": "RELEASE"}
|
||||
logger.info("TRANSPORT RELEASE: POST %s corrNr=%s", url, corr_nr)
|
||||
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
||||
logger.info("TRANSPORT RELEASE RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
logger.error("TRANSPORT RELEASE ERROR: %s", resp.text[:2000])
|
||||
raise SapCliError(
|
||||
f"释放传输请求失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
return True
|
||||
|
||||
def transport_objects(self, corr_nr: str) -> list[dict[str, str]]:
|
||||
"""列出传输请求中的对象。
|
||||
|
||||
Args:
|
||||
corr_nr: 传输请求编号。
|
||||
|
||||
Returns:
|
||||
列表，每项含 ``name``, ``type``。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/cts/transportrequests/{corr_nr}"
|
||||
hdrs: dict[str, str] = {
|
||||
**self._headers(),
|
||||
"Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
|
||||
}
|
||||
params: dict[str, str] = {"withObjects": "true"}
|
||||
logger.info("TRANSPORT OBJECTS: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs, params=params)
|
||||
logger.info("TRANSPORT OBJECTS RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
|
||||
root = ET.fromstring(resp.content)
|
||||
results: list[dict[str, str]] = []
|
||||
core_ns = "http://www.sap.com/adt/core"
|
||||
for ref in root.findall(f".//{{{core_ns}}}objectReference"):
|
||||
obj_name = ref.attrib.get(f"{{{core_ns}}}name", "")
|
||||
obj_type = ref.attrib.get(f"{{{core_ns}}}type", "")
|
||||
results.append({"name": obj_name, "type": obj_type})
|
||||
return results
|
||||
Reference in New Issue
Block a user