Thank you for sharing!
I’ve created sample document here:
[Gen.ledger - Grist]
Add a record or modify existing one in Transactions table. Every minute external script (herebelow) checks for change and updates table Entries (of course, in real app this period shall be much lower). I will leave it running on my machine for a while. It is of course very questionable approach as a lot of code resides outside of the grist document.
#!/usr/bin/env python3
from grist_api import GristDocAPI
import os
from datetime import datetime as dt
from random import choice
from dataclasses import dataclass
from random import random, sample
import logging, time
class MyEx(Exception):
pass
class GristDoc():
def __init__(self, updater, logfile = None,
server=None, doc_id=None, api_key=None):
''' sets connection and
Args:
server: txt, url, e.g https://something.com
doc_id: txt, get it from grist server
api_key: txt, generated by grist server
updater: object with compulsory methods
check_updated
on_update
'''
self.logger = logging.getLogger(__name__)
logging.basicConfig(
format='%(asctime)s %(message)s',
encoding='utf-8',
level = logging.DEBUG)
if not updater or not hasattr(updater, 'fetch_updated') \
or not hasattr(updater, 'on_update'):
raise MyEx("no suitable updater instance")
self.updater = updater
self.server = server
self.doc_id = doc_id
self.api_key = api_key
self.update_conn_data()
self.api = GristDocAPI(self.doc_id, server = self.server)
def update_conn_data(self):
'''fetch connection data, if missing
Args:
'''
for a in ['server','doc_id','api_key']:
if not getattr(self, a):
#attempt to fetch value from environment var
env_var = f'GRIST_{a.upper()}'
try:
val = os.environ[env_var]
except (KeyError):
val = ''
if not val:
raise MyEx(f"attribute {a} is not defined!\nEither provide it as argument to GristDoc class or set environment variable {env_var}")
setattr(self, a, val)
def check_for_updates(self):
'''
'''
self.updater.fetch_updated(api = self.api)
self.logger.info(f'{len(self.updater.updated)} records updated')
def action_on_update(self):
'''
'''
self.updater.on_update(api = self.api)
self.logger.info(f'finished action stage')
class UpdaterTransaction():
def __init__(self):
pass
def fetch_updated(self, api):
# self.updated = api.call('tables/{}/records?filter={"{}":[true]}'.\
# format(self.check_table, self.check_column))
self.updated = api.fetch_table('Transactions',
{"ext_check": True})
def on_update(self, api):
''' generate/update records
'''
for r in self.updated:
# this code is for demonstration purpose only
# just to illustrate the extend the algorithm taken away
# from the grist
amounts = []
if r.narration:
amounts = r.narration.split()
comment = "amount from the transaction's narration"
if not amounts:
comment = "generated amount"
amounts = []
for i in range(2):
amounts.append(round(random() * 10000 / (12 + i),2))
amounts.append(-round(sum(amounts),2))
accounts = api.fetch_table('Accounts')
accounts_rand = sample(range(len(accounts)), 3)
#create 3 sample transactions
new_records = []
for i in range(3):
new_records.append(
{'transaction': r.id,
'account': accounts_rand[i],
'amount': amounts[i],
'comment': comment
}
)
api.add_records(table_name = 'Entries',
record_dicts = new_records)
#mark "parent" record as "done"
#in real life we shall mark "parent" record as done much promptly
api.update_records(table_name = 'Transactions',
record_dicts = [ {"id": r.id, "ext_check": False} \
for r in self.updated ])
self.updated = []
def main():
d = GristDoc(UpdaterTransaction())
while True:
d.check_for_updates()
d.action_on_update()
#should be updated much often, of course
time.sleep(42)
if __name__ == "__main__":
main()