Method
create
(
self,
*,
name: str,
cron_expr: str,
command: str,
fleet_id: str | None = None,
description: str = "",
timezone: str = "UTC",
target_kind: str = "all_fleet_machines",
is_enabled: bool = True,
machine_ids: Sequence[str] | None = None,
)
Source from the content-addressed store, hash-verified
| 49 | return (await self._unary(req)).schedule_view |
| 50 | |
| 51 | async def create( |
| 52 | self, |
| 53 | *, |
| 54 | name: str, |
| 55 | cron_expr: str, |
| 56 | command: str, |
| 57 | fleet_id: str | None = None, |
| 58 | description: str = "", |
| 59 | timezone: str = "UTC", |
| 60 | target_kind: str = "all_fleet_machines", |
| 61 | is_enabled: bool = True, |
| 62 | machine_ids: Sequence[str] | None = None, |
| 63 | ) -> ScheduleView: |
| 64 | fid = self._fleet(fleet_id) |
| 65 | req = pb.Envelope( |
| 66 | create_schedule_req=s_pb.CreateScheduleRequest( |
| 67 | fleet_id=fid, |
| 68 | name=name, |
| 69 | cron_expr=cron_expr, |
| 70 | command=command, |
| 71 | description=description, |
| 72 | timezone=timezone, |
| 73 | target_kind=_TARGET.get(target_kind, common_pb.SCHEDULE_TARGET_KIND_ALL_FLEET_MACHINES), |
| 74 | is_enabled=is_enabled, |
| 75 | machine_ids=[str(x) for x in (machine_ids or [])], |
| 76 | ) |
| 77 | ) |
| 78 | return (await self._unary(req)).schedule_view |
| 79 | |
| 80 | async def update( |
| 81 | self, |
Callers
nothing calls this directly
Tested by
no test coverage detected