import requests from decimal import getcontext, Decimal, ROUND_HALF_UP from datetime import datetime from trytond.model import fields from trytond.model import (ModelSQL, ModelView) from trytond.pool import Pool, PoolMeta import logging logger = logging.getLogger(__name__) class Cron(metaclass=PoolMeta): __name__ = 'ir.cron' @classmethod def __setup__(cls): super().__setup__() cls.method.selection.append( ('forex.cron|update_forex', "Update Forex Prices")) class PriceCron(ModelSQL, ModelView): "Price Cron" __name__ = 'forex.cron' frequency = fields.Selection([ ('daily', "Daily"), ('weekly', "Weekly"), ('monthly', "Monthly"), ], "Frequency", required=True, help="How frequently rates must be updated.") last_update = fields.Date("Last Update", required=True) @classmethod def __setup__(cls): super().__setup__() cls._buttons.update({ 'run': {}, }) @classmethod def default_frequency(cls): return 'daily' @classmethod def default_last_update(cls): pool = Pool() Date = pool.get('ir.date') return Date.today() @classmethod @ModelView.button def run(cls, crons): cls.update_forex(crons) @classmethod def update_forex(cls): Currency = Pool().get('currency.currency') Rate = Pool().get('currency.currency.rate') # On suppose que l'EUR existe déjà dans la base eur, = Currency.search([('name', '=', 'EUR')]) # Appel API Frankfurter url = "https://api.frankfurter.app/latest" params = {"base": "EUR", "symbols": "USD"} resp = requests.get(url, params=params) resp.raise_for_status() data = resp.json() rate_value = round(Decimal(data["rates"]["USD"]),4) rate_date = datetime.strptime(data["date"], "%Y-%m-%d").date() logger.info(f"Taux EUR/USD : {rate_value} ({rate_date})") # Vérifie si un taux existe déjà pour ce jour existing = Rate.search([ ('currency', '=', eur.id), ('date', '=', rate_date), ]) if not existing and rate_value: rate = Rate(currency=eur, rate=round(1/rate_value,6), date=rate_date) rate.save() logger.info(f"Taux EUR/USD ajouté : {rate_value} ({rate_date})") else: logger.info(f"Taux déjà existant pour {rate_date}")