How to use the kopf.on.field function in kopf

To help you get started, we’ve selected a few kopf examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github zalando-incubator / kopf / tests / registries / test_decorators.py View on Github external
        @kopf.on.field('group', 'version', 'plural')
        def fn(**_):
            pass
github elemental-lf / benji / src / benji / k8s_operator / crd / operator_config.py View on Github external
@kopf.on.field('batch', 'v1', 'jobs', field='status', labels={LABEL_PARENT_KIND: CRD_OPERATOR_CONFIG.name})
def benji_track_job_status_maintenance(**_) -> Optional[Dict[str, Any]]:
    return track_job_status(crd=CRD_OPERATOR_CONFIG, **_)
github zalando-incubator / kopf / examples / 05-handlers / example.py View on Github external
@kopf.on.field('zalando.org', 'v1', 'kopfexamples', field='spec.field')
def field_fn(old, new, **kwargs):
    print(f'FIELD CHANGED: {old} -> {new}')
github elemental-lf / benji / src / benji / k8s_operator / crd / retention_schedule.py View on Github external
@kopf.on.field('batch', 'v1', 'jobs', field='status', labels={LABEL_PARENT_KIND: ClusterBenjiRetentionSchedule.kind})
def benji_track_job_status_cluster_retention_schedule(**kwargs) -> Optional[Dict[str, Any]]:
    return track_job_status(crd=ClusterBenjiRetentionSchedule, **kwargs)
github zalando-incubator / kopf / examples / 99-all-at-once / example.py View on Github external
@kopf.on.field('zalando.org', 'v1', 'kopfexamples', field='spec.lst')
def update_lst(body, meta, spec, status, old, new, **kwargs):
    print(f'Handling the FIELD = {old} -> {new}')
github elemental-lf / benji / src / benji / k8s_operator / crd / retention_schedule.py View on Github external
@kopf.on.field('batch', 'v1', 'jobs', field='status', labels={LABEL_PARENT_KIND: BenjiRetentionSchedule.kind})
def benji_track_job_status_retention_schedule(**kwargs) -> Optional[Dict[str, Any]]:
    return track_job_status(crd=BenjiRetentionSchedule, **kwargs)
github elemental-lf / benji / src / benji / k8s_operator / crd / version.py View on Github external
@kopf.on.field(*BenjiVersion.group_version_plural(), field='status.protected')
def benji_protect(name: str, status: Dict[str, Any], body: Dict[str, Any], **_) -> Optional[Dict[str, Any]]:
    benji = APIClient()
    check_version_access(benji, name, body)
    protected = status.get('protected', False)
    benji.protect(name, protected)