cancel
Showing results for 
Search instead for 
Did you mean: 
cancel
1108
Views
0
Helpful
3
Replies

Validation Callback in Python

erdemk
Level 1
Level 1

Dear NSO Community,

 

Could some one give me a working example or direct me to the right source..

 

I want to try using Validation Callback with python. In development guide, there is info with Java. In the following thread, there is actually very detailed info on how to use Validation Callback with Python. But as

far as i understand , there is some modification on the Python API. As thread is dated back to 2017 and i am using NSO5.3.

 

Thread explaining Validation Callback with Python: https://community.cisco.com/t5/nso-developer-hub-discussions/validation-callback-in-python/td-p/3461522

 

I tried "trans_set_fd(tctx, self.daemon.wsock" instead of "trans_set_fd(tctx, self.daemon._wsock" which is in the original post  :

def cb_init(self, tctx):
    self.maapi.attach(tctx)
    trans_set_fd(tctx, self.daemon._wsock)

 

NSO tells me this :

'wsock no longer supported. Daemon now use multiple workers. '
Exception: wsock no longer supported. Daemon now use multiple workers. See dp.take_worker_socket and dp.return_worker_socket.

 

Thanks in advance.

1 Accepted Solution

Accepted Solutions

gmuloche
Cisco Employee
Cisco Employee

Hello,

 

the following changes work for me - the wsock must be taken from a Pool and returned at the end:

 

# -*- mode: python; python-indent: 4 -*-
import ncs
from ncs.dp import Daemon
from ncs.maapi import Maapi
from _ncs.dp import register_valpoint_cb, register_trans_validate_cb, trans_set_fd


class Validation(object):

    def __init__(self, log):
        self.log = log
        self.daemon = Daemon("foo-validation-daemon", log=log)
        self._state = Daemon.State(self.daemon)
        self.maapi = Maapi()                                                                                                                                                                                                                          register_trans_validate_cb(self.daemon.ctx(), self)
        register_valpoint_cb(self.daemon.ctx(),
                             'the-validation-callpoint-name',
                             self)
        self.daemon.start()

    def cb_validate(self, tctx, kp, newval):
        self.log.info('validate called')
        # return ncs.CONFD_ERR to fail the validation otherwise
        # ncs.CONFD_OK
        return ncs.CONFD_OK

    def cb_init(self, tctx):
        self.maapi.attach(tctx)
        name = 'th-{0}'.format(tctx.th)
        wsock = ncs.dp.take_worker_socket(self._state, name, self._make_key(tctx))

        # Associate the worker socket with the transaction
        trans_set_fd(tctx, wsock)

    def cb_stop(self, tctx):
        ncs.dp.return_worker_socket(self._state, self._make_key(tctx))
        self.maapi.detach(tctx)

    def _make_key(self, tctx):
        return '{0}-{1}'.format(id(self), tctx.th)


class Main(ncs.application.Application):
    def setup(self):
        self.log.info('Main RUNNING')
        self.v = Validation(self.log)

    def teardown(self):
        self.log.info('Main FINISHED')

View solution in original post

3 Replies 3

gmuloche
Cisco Employee
Cisco Employee

Hello,

 

the following changes work for me - the wsock must be taken from a Pool and returned at the end:

 

# -*- mode: python; python-indent: 4 -*-
import ncs
from ncs.dp import Daemon
from ncs.maapi import Maapi
from _ncs.dp import register_valpoint_cb, register_trans_validate_cb, trans_set_fd


class Validation(object):

    def __init__(self, log):
        self.log = log
        self.daemon = Daemon("foo-validation-daemon", log=log)
        self._state = Daemon.State(self.daemon)
        self.maapi = Maapi()                                                                                                                                                                                                                          register_trans_validate_cb(self.daemon.ctx(), self)
        register_valpoint_cb(self.daemon.ctx(),
                             'the-validation-callpoint-name',
                             self)
        self.daemon.start()

    def cb_validate(self, tctx, kp, newval):
        self.log.info('validate called')
        # return ncs.CONFD_ERR to fail the validation otherwise
        # ncs.CONFD_OK
        return ncs.CONFD_OK

    def cb_init(self, tctx):
        self.maapi.attach(tctx)
        name = 'th-{0}'.format(tctx.th)
        wsock = ncs.dp.take_worker_socket(self._state, name, self._make_key(tctx))

        # Associate the worker socket with the transaction
        trans_set_fd(tctx, wsock)

    def cb_stop(self, tctx):
        ncs.dp.return_worker_socket(self._state, self._make_key(tctx))
        self.maapi.detach(tctx)

    def _make_key(self, tctx):
        return '{0}-{1}'.format(id(self), tctx.th)


class Main(ncs.application.Application):
    def setup(self):
        self.log.info('Main RUNNING')
        self.v = Validation(self.log)

    def teardown(self):
        self.log.info('Main FINISHED')

This also worked for me. Many thanks.

One more question on this topic, if possible.. Can i deploy 2 validation points from different points in yang model, within same service ..

Can someone put an example code, how to validate the input in cb_validate? The service callback is well documented, and understood, but this one has no documentation at all. For example I'd like to walk a configuration object tree (but how to get access to the tree, when root is not passed to the callback), and compare that to the scratchpad configuration (which is in newval, I guess).

TIA,
robert
Getting Started

Find answers to your questions by entering keywords or phrases in the Search bar above. New here? Use these resources to familiarize yourself with the NSO Developer community: