core

core

DomoObjective dataclass

DomoObjective(
    auth: DomoAuth,
    id: int,
    raw: dict,
    name: str = None,
    description: str = None,
    starts_at: Any = None,
    expires_at: Any = None,
    status: str = None,
    owner_id: int = None,
    owners: list[Any] = None,
    assignees: list[Any] = None,
    period_id: int = None,
    parent_id: int = None,
    tags: list[Any] = None,
    write_access: bool = None,
    type: str = None,
)

Bases: DomoEntity

Represents a Domo Goal/Objective.

create async classmethod

create(
    auth: DomoAuth,
    name: str,
    period_id: int,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjective

Create a new objective.

Source code in src/crew_dcs/classes/DomoObjective/core.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    name: str,
    period_id: int,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjective:
    """Create a new objective."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.create_objective(
        auth=auth,
        name=name,
        period_id=period_id,
        context=context,
    )

    if return_raw:
        return res

    return cls.from_dict(auth=auth, obj=res.response)

delete async

delete(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Delete this objective.

Source code in src/crew_dcs/classes/DomoObjective/core.py

async def delete(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete this objective."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await objectives_routes.delete_objective(
        auth=self.auth,
        objective_id=self.id,
        context=context,
    )

from_dict classmethod

from_dict(
    auth: DomoAuth, obj: dict[str, Any], **kwargs
) -> DomoObjective

Converts an objectives API response into a DomoObjective instance.

Source code in src/crew_dcs/classes/DomoObjective/core.py

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any], **kwargs) -> DomoObjective:
    """Converts an objectives API response into a DomoObjective instance."""

    return cls(
        id=obj.get("id"),
        auth=auth,
        name=obj.get("name"),
        description=obj.get("description"),
        starts_at=cd.convert_string_to_datetime(obj.get("startsAt")),
        expires_at=cd.convert_string_to_datetime(obj.get("expiresAt")),
        status=obj.get("status"),
        owner_id=obj.get("ownerId"),
        owners=obj.get("owners"),
        assignees=obj.get("assignees"),
        period_id=obj.get("periodId"),
        parent_id=obj.get("parentId"),
        tags=obj.get("tags"),
        write_access=obj.get("writeAccess"),
        type=obj.get("type"),
        raw=obj,
        **kwargs,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    objective_id: int | str,
    period_id: int | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjective

Retrieve a single objective by ID.

The Goals API does not expose a dedicated "get objective by id" endpoint, so this searches all objectives and matches by ID.

Source code in src/crew_dcs/classes/DomoObjective/core.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    objective_id: int | str,
    period_id: int | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjective:
    """Retrieve a single objective by ID.

    The Goals API does not expose a dedicated "get objective by id"
    endpoint, so this searches all objectives and matches by ID.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.get_objectives(
        auth=auth,
        period_id=period_id,
        context=context,
    )

    if return_raw:
        return res

    for obj in res.response:
        if str(obj.get("id")) == str(objective_id):
            return cls.from_dict(auth=auth, obj=obj)

    raise DomoObjective_NoMatchError(
        objective_id=objective_id, domo_instance=auth.domo_instance
    )

get_entity_by_id async classmethod

get_entity_by_id(entity_id, **kwargs) -> DomoObjective

Alias for get_by_id.

Source code in src/crew_dcs/classes/DomoObjective/core.py

@classmethod
async def get_entity_by_id(cls, entity_id, **kwargs) -> DomoObjective:
    """Alias for get_by_id."""
    return await cls.get_by_id(objective_id=entity_id, **kwargs)

update async

update(
    payload: dict | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjective

Update this objective.

Source code in src/crew_dcs/classes/DomoObjective/core.py

async def update(
    self,
    payload: dict | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjective:
    """Update this objective."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    payload = payload or objectives_routes.generate_create_objective_body(
        name=self.name,
        period_id=self.period_id,
        description=self.description,
        status=self.status,
        owner_id=self.owner_id,
        owners=self.owners,
        assignees=self.assignees,
        parent_id=self.parent_id,
        tags=self.tags,
        write_access=self.write_access,
        type=self.type,
    )

    res = await objectives_routes.update_objective(
        auth=self.auth,
        objective_id=self.id,
        payload=payload,
        period_id=self.period_id,
        context=context,
    )

    if return_raw:
        return res

    updated = self.from_dict(auth=self.auth, obj=res.response)
    self.__dict__.update(
        {
            k: v
            for k, v in updated.__dict__.items()
            if k not in ("KeyResults", "Tags")
        }
    )
    return self

DomoObjective_NoMatchError

DomoObjective_NoMatchError(
    objective_id, domo_instance, **kwargs
)

Bases: ClassError

Raised when a specific objective cannot be found.

Source code in src/crew_dcs/classes/DomoObjective/core.py

def __init__(self, objective_id, domo_instance, **kwargs):
    super().__init__(
        message=f"Objective '{objective_id}' not found",
        entity_id=objective_id,
        domo_instance=domo_instance,
        **kwargs,
    )

DomoObjectives dataclass

DomoObjectives(
    auth: DomoAuth, objectives: list[DomoObjective] = None
)

Bases: DomoManager

Manager for retrieving collections of objectives.

get async

get(
    period_id: int | None = None,
    filter_key_results: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoObjective]

Search for objectives.

Source code in src/crew_dcs/classes/DomoObjective/core.py

async def get(
    self,
    period_id: int | None = None,
    filter_key_results: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoObjective]:
    """Search for objectives."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.get_objectives(
        auth=self.auth,
        period_id=period_id,
        filter_key_results=filter_key_results,
        context=context,
    )

    if return_raw:
        return res

    self.objectives = [
        DomoObjective.from_dict(auth=self.auth, obj=obj) for obj in res.response
    ]
    return self.objectives