from argparse import _SubParsersAction
from functools import partial
from typing import Optional
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.hq_params.stored_procedure import ParamsBuilder
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class RestrictedEntitlementModeRow(TableRow):
# object name: type hinting = "column name in the db response"
ProfileId: int
ProfileName: str
[docs]
class RestrictedEntitlementMode(DbObject):
"""
DbObject for managing Restricted Entitlement Mode (REM) profile configuration.
REM is a security feature that restricts customer access based on configurable profiles.
Each profile defines which groups can use it, which transactions are allowed,
and which audit actions apply.
"""
NAME = "RestrictedEntitlementMode"
REPRESENTATION_ROW_CLASS = RestrictedEntitlementModeRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
# get_profiles command
profiles_parser = parser.add_parser(
"get_rem_profiles", help="Get REM profiles (all or filtered by GroupID)"
)
profiles_parser.set_defaults(parser="get_profiles")
profiles_parser.set_defaults(
func=partial(self.get_profiles, serialize_for_cli=True)
)
profiles_parser.add_argument(
"--group-id", type=int, dest="group_id", help="Optional GroupID filter"
)
# get_profile_groups command
groups_parser = parser.add_parser(
"get_rem_profile_groups", help="Get groups that can use a REM profile"
)
groups_parser.set_defaults(parser="get_profile_groups")
groups_parser.set_defaults(
func=partial(self.get_profile_groups, serialize_for_cli=True)
)
groups_parser.add_argument("profile_id", type=int, help="Profile ID")
# get_profile_transactions command
trans_parser = parser.add_parser(
"get_rem_profile_transactions",
help="Get transaction configuration for a REM profile",
)
trans_parser.set_defaults(parser="get_profile_transactions")
trans_parser.set_defaults(
func=partial(self.get_profile_transactions, serialize_for_cli=True)
)
trans_parser.add_argument("profile_id", type=int, help="Profile ID")
# get_profile_audit_actions command
audit_parser = parser.add_parser(
"get_rem_profile_audit_actions",
help="Get audit action configuration for a REM profile",
)
audit_parser.set_defaults(parser="get_profile_audit_actions")
audit_parser.set_defaults(
func=partial(self.get_profile_audit_actions, serialize_for_cli=True)
)
audit_parser.add_argument("profile_id", type=int, help="Profile ID")
[docs]
async def get_profiles(
self, group_id: Optional[int] = None, serialize_for_cli: bool = False
) -> list[dict]:
"""
Get REM profiles, optionally filtered by GroupID.
:param group_id: Optional GroupID to filter profiles by
:param serialize_for_cli: Serialize the response for CLI output
:return: List of profile dictionaries with ProfileId and ProfileName
"""
if group_id is not None and group_id <= 0:
raise ValueError("group_id must be a positive integer")
params_builder = ParamsBuilder()
if group_id is not None:
params_builder.add_param(D_TYPES.Int, "groupId", group_id)
response = await self.call_hq(
"sdk_GetREMProfiles",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
if serialize_for_cli:
columns = ["ProfileId", "ProfileName"]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_profile_groups(
self, profile_id: int, serialize_for_cli: bool = False
) -> list[dict]:
"""
Get group mappings for a REM profile.
:param profile_id: Profile ID to get groups for
:param serialize_for_cli: Serialize the response for CLI output
:return: List of group dictionaries with GroupID, GroupName, ProfileId, and ProfileName
"""
if profile_id <= 0:
raise ValueError("profile_id must be a positive integer")
params_builder = ParamsBuilder()
params_builder.add_param(D_TYPES.Int, "profileId", profile_id)
response = await self.call_hq(
"sdk_GetREMProfileGroups",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
# Sort client-side to reduce SQL Server load
response = sorted(response, key=lambda x: x.findtext("GroupName", ""))
if serialize_for_cli:
columns = ["GroupID", "GroupName", "ProfileId", "ProfileName"]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_profile_transactions(
self, profile_id: int, serialize_for_cli: bool = False
) -> list[dict]:
"""
Get transaction configuration for a REM profile.
:param profile_id: Profile ID to get transactions for
:param serialize_for_cli: Serialize the response for CLI output
:return: List of transaction configuration dictionaries with all transaction fields,
TransactionTypeName, TransactionTypeDescription, and ProfileName
"""
if profile_id <= 0:
raise ValueError("profile_id must be a positive integer")
params_builder = ParamsBuilder()
params_builder.add_param(D_TYPES.Int, "profileId", profile_id)
response = await self.call_hq(
"sdk_GetREMProfileTransactions",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
# Sort client-side to reduce SQL Server load
response = sorted(
response,
key=lambda x: (
x.findtext("TransactionTypeName", ""),
x.findtext("GtFlavorShortName", ""),
),
)
if serialize_for_cli:
# Include key transaction config fields from Q2_RestrictedEntitlementTransaction
# Grouped logically: identification, permissions, monetary limits, count limits, special limits
columns = [
# Transaction and profile identification
"TransactionTypeID",
"TransactionTypeName",
"ProfileId",
"ProfileName",
# Permission flags
"Enabled",
"Authorize",
"Cancel",
"Create",
"CreateRestricted",
"Update",
"View",
# Monetary limits
"LimitPerTransaction",
"LimitPerDay",
"LimitPerMonth",
# Count limits
"CountPerDay",
"CountPerMonth",
# Special approval and token limits
"DualApprovalLimit",
"TokenRequiredLimit",
"DraftLimit",
# Generated transaction flavor
"GtFlavorShortName",
"GtFlavorDescription",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_profile_audit_actions(
self, profile_id: int, serialize_for_cli: bool = False
) -> list[dict]:
"""
Get audit action configuration for a REM profile.
:param profile_id: Profile ID to get audit actions for
:param serialize_for_cli: Serialize the response for CLI output
:return: List of audit action configuration dictionaries with ActionID, ActionName,
ActionDescription, Allow, and ProfileName
"""
if profile_id <= 0:
raise ValueError("profile_id must be a positive integer")
params_builder = ParamsBuilder()
params_builder.add_param(D_TYPES.Int, "profileId", profile_id)
response = await self.call_hq(
"sdk_GetREMProfileAuditActions",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
# Sort client-side to reduce SQL Server load
response = sorted(response, key=lambda x: x.findtext("ActionName", ""))
if serialize_for_cli:
columns = [
"ActionID",
"ActionName",
"ActionDescription",
"Allow",
"ProfileId",
"ProfileName",
]
response = self.serialize_for_cli(response, columns)
return response