Daml dit if Timed Events

Hello all,

I was reading the documentation on daml dit if regarding IntegrationTimeEvents, and it left me wondering if it would be possible to leverage it to schedule events at a fixed point in time, for example, running the event every day at midnight.
If not how would you advise me to best use this library (or others) to achieve this.

Thank you in advance.

The only native support for timing within daml-dit-if is through periodic timed events.

https://github.com/digital-asset/daml-dit-if/blob/7a529fc90d75d3f6f53bce748d0b52749f682291/daml_dit_if/api/__init__.py#L70

This decorator requests that the framework issue a request to run the associated event handler at the specified interval. No strong guarantees are made about exact timing.

However, this is where daml-dit-if's support for internal queuing comes into play:

https://github.com/digital-asset/daml-dit-if/blob/7a529fc90d75d3f6f53bce748d0b52749f682291/daml_dit_if/api/__init__.py#L55

One of the primary motivations behind queuing is to allow daml-dit-if to support integrations with components for which it has no direct support. As you mention in the Slack thread, there are libraries like aioschedule that do provide more time based scheduling support.

PyPI

aioschedule

Job scheduling for humans.

Queues can enable you to use these. The basic pattern to follow is the same pattern used in the Timer integration within the core pack:

github.com

digital-asset/daml-dit-integration-core/blob/7b50d3af80ee93f1289a1e6594b2ff7bbffbe5d4/src/core_int/integration_timer.py#L48

      
        

            
async def on_contract_created(event):
            
    LOG.debug('Created CID: %r', event.cid)
            
    active_cids.add(event.cid)
            

            
@events.ledger.contract_archived(env.targetTemplate)
            
async def on_ledger_archived(event):
            
    LOG.debug('Archived CID: %r', event.cid)
            
    active_cids.discard(event.cid)
            

            
@events.time.periodic_interval(env.interval, label='Periodic Timer')
            
async def interval_timer_elapsed():
            
    LOG.debug('Timer elapsed: %r', active_cids)
            
    commands = [exercise(cid, env.templateChoice, {})
            
                for cid
            
                in active_cids]
            

            
    for batch in batch_commands(commands, env.batchSize):
            
        await env.queue.put(batch)
            

            
@events.queue.message()
            
async def handle_command_batch(message):
        
      
    

There is one event handler for a queue event, and another process that drops events on the queue at the appropriate time. For the timer integration linked above, the timing the timing is managed by the interval timer built into daml-dit-if, but there is nothing that would prevent the use of aioschedule or similar.

Another aspect that you’ll need to be aware of is that for aioschedule to work, it appears to need to run an asyncio coroutine. These can be passed to daml-dit-if to run by returning them from the entrypoint function:

There’s an example of how that works in the Exberry integration (which is designed with a background coroutine to manage a connection):

github.com

digital-asset/daml-dit-integration-exberry/blob/5a490df3f4e51c0402642a39a55b807c740524c3/src/exberry_int/main.py#L198

      
        

            
    elif 'errorType' in msg:
            
        msg_data = msg['d']
            
        LOG.info(f'Received mass cancel failure, creating {EXBERRY.MassCancelSuccess}')
            
        return create(EXBERRY.MassCancelFailure, {
            
            'integrationParty': env.party,
            
            'sid': msg['sid'],
            
            'errorCode': msg_data['errorCode'],
            
            'errorMessage': msg_data['errorMessage'],
            
        })
            

            
return integration.connect()
        
      
    

Hello @Michael_Schaeffer, the indications you have provided have indeed been helpful, but now I’m facing a different issue, which is, when starting the integration first the first time (so the .ddit-venv hasn’t been created yet), I get the following error regarding the aioschedule module.

ERROR: Could not find a version that satisfies the requirement aioschedule==0.5.2 (from -r requirements.txt (line 2)) (from versions: none)
ERROR: No matching distribution found for aioschedule==0.5.2 (from -r requirements.txt (line 2))

As a quick fix I just go inside the env and install it manually, but that won’t work to start it on docker containers, for example. Do you know of a way to fix this?

Thanks