Skip to content

activity_log

activity_log

ActivityLog_GET_Error 

ActivityLog_GET_Error(
    message: str | None = None, res=None, **kwargs
)

Bases: RouteError

Raised when activity log retrieval operations fail.

Source code in src/crew_dcs/routes/activity_log.py 
30
31
32
33
34
35
def __init__(self, message: str | None = None, res=None, **kwargs):
    super().__init__(
        message=message or "Activity log retrieval failed",
        res=res,
        **kwargs,
    )

get_activity_log_object_types async 

get_activity_log_object_types(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

retrieves a list of valid objectTypes that can be used to search the activity_log API

Source code in src/crew_dcs/routes/activity_log.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_activity_log_object_types(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """retrieves a list of valid objectTypes that can be used to search the activity_log API"""

    url = f"https://{auth.domo_instance}.domo.com/api/audit/v1/user-audits/objectTypes"

    res = await gd.get_data(
        url=url,
        method="GET",
        auth=auth,
        context=context,
    )

    if not res.is_success:
        raise ActivityLog_GET_Error(
            message="Failed to get activity log object types", res=res
        )

    return res

search_activity_log async 

search_activity_log(
    auth: DomoAuth,
    start_time: int,
    end_time: int,
    maximum: int | None = None,
    object_type: str | None = None,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

loops over activity log api to retrieve audit logs

Source code in src/crew_dcs/routes/activity_log.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_activity_log(
    auth: DomoAuth,
    start_time: int,  # epoch time in milliseconds
    end_time: int,  # epoch time in milliseconds
    maximum: int | None = None,
    object_type: str | None = None,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """loops over activity log api to retrieve audit logs"""

    url = f"https://{auth.domo_instance}.domo.com/api/audit/v1/user-audits"

    if object_type and object_type != "ACTIVITY_LOG":
        url = f"{url}/objectTypes/{object_type}"

    fixed_params = {"end": end_time, "start": start_time}

    offset_params = {
        "offset": "offset",
        "limit": "limit",
    }

    def arr_fn(res) -> list[dict]:
        return res.response

    res = await gd.looper(
        auth=auth,
        method="GET",
        url=url,
        arr_fn=arr_fn,
        fixed_params=fixed_params,
        offset_params=offset_params,
        maximum=maximum,
        limit=1000,
        debug_loop=debug_loop,
        context=context,
    )

    if not res.is_success:
        raise ActivityLog_GET_Error(res=res)

    return res