[PITCH+DESAFIO] Cronzun, um módulo para cron nem tão diferente de outros

Galera, resolvi trazer este pitch+desafio, porque acho muito legal a interação da comunidade do TN! Criei um projeto no Github como um desafio pra mim com a implementação de um processador de expressões cron para prever a próxima data que o cron será executado.

O desafio aqui é:

• Como que você melhoraria o código?

Porém, o desafio não é apenas isso, se fosse, seria bem fácil só pesquisando no ChatGPT!

Mas a ideia não é pesquisar nem em ChatGPT (ou algum Chat de IA) e também não vale utilizar algum framework (Como o Croniter do Python) para melhorar a implementação deste código.

Quanto mais puro e simples o código ficar (com 1 arquivo só de implementação e 1 arquivo só de testes) melhor será essa implementação e as melhorias.

Na verdade este projeto é um desafio para meus conhecimentos de desenvolvimento, desafio para entender como este processamento de datas de cron funciona, ou seja, ao abrir o site do CronMaker, fiquei muito tentado a entender o funcionamento por trás da compreensão da expressão cron e a disponibilização das próximas datas de execução. Isso foi lindo de se ver, mas não tão lindo de se implementar.

Sei que o código tem várias melhorias, mas até aqui acredito que consegui bastante insights legais.

Para a implementação deste código utilizei a estratégia de TDD, ou seja, inicio criando um teste com uma regra simples, executo o teste esperando a falha, implemento o código que satisfaça o teste, executo o teste esperando o sucesso TOTAL, ou seja, no começo as falhas em outras partes do código serão raras, porém com o passar do tempo, as falhas começarão a aparecer pois quando você altera um trecho de um código, outro trecho poderá ser impactado, alguma regra de negócio importante poderá ser alterada e não é isso que queremos.

Queremos um código de qualidade (neste meu caso não tem qualidade aparente), mas tem uma certa qualidade de entrega de valor no seu objetivo esperado.

Este é o código até o momento no arquivo cronzun.py.

import typing
import datetime

# https://stackoverflow.com/questions/4610904/calculate-next-scheduled-time-based-on-cron-spec
# https://www.geeksforgeeks.org/writing-cron-expressions-for-scheduling-tasks/
# https://crontab.cronhub.io/
# https://freeformatter.com/cron-expression-generator-quartz.html

class Cronzun:
    cron_expression: str = None
    from_datetime: typing.Union[datetime.datetime, None] = None

    def __init__(self, cron_expression: str, 
                 from_datetime: typing.Union[datetime.datetime, None] = None) -> None:
        # cron_expression = '* * * * * ? *'
        self.cron_expression = cron_expression
        self.from_datetime = self.default(from_datetime, datetime.datetime.now())

    def default(self, value_verified, value_default) -> typing.Union[typing.Any, None]:
        return value_verified if value_verified else value_default

    def set_from_datetime(self, dt: datetime.datetime) -> None:
        self.from_datetime = dt

    def get_next_date(self) -> datetime.datetime:
        cron_expr =      self.cron_expression.split(' ')
        _seconds =       cron_expr[0]
        _minutes =       cron_expr[1]
        _hours =         cron_expr[2]
        _day_of_month =  cron_expr[3]
        _month =         cron_expr[4]
        _day_of_week =   cron_expr[5]
        _year =          cron_expr[6]
    
        from_dt = self.from_datetime
        next_dt = from_dt

        if _seconds == '*':
            seconds = from_dt.second + 1
            seconds = 0 if seconds == 60 else seconds
        else:
            seconds = int(_seconds)
            if not (seconds >= 0 and seconds <= 59):
                raise Exception('Second value must be between 0 and 59!')

        next_dt = datetime.datetime(
            next_dt.year, next_dt.month, next_dt.day,
            next_dt.hour, next_dt.minute, seconds
        )

        # Minutes
        if _seconds != '*' and _minutes == '*':
            if from_dt.second > seconds:
                minutes = from_dt.minute + 1
                minutes = 0 if minutes == 60 else minutes
            elif from_dt.second <= seconds:
                minutes = from_dt.minute
        elif _seconds == '*' and _minutes == '*':
            minutes = from_dt.minute
        else:
            minutes = int(_minutes)
            if not (minutes >= 0 and minutes <= 59):
                raise Exception('Minute value must be between 0 and 59!')

        next_dt = datetime.datetime(
            next_dt.year, next_dt.month, next_dt.day,
            next_dt.hour, minutes, seconds
        )

        # Hours
        if  _minutes != '*' and _hours == '*':
            hours = from_dt.hour + 1
            hours = 0 if hours == 24 else hours
        elif _minutes == '*' and _hours == '*':
            hours = from_dt.hour
        else:
            hours = int(_hours)
            if not (hours >= 0 and hours <= 23):
                raise Exception('Hour value must be between 0 and 23!')

        next_dt = datetime.datetime(
            next_dt.year, next_dt.month, next_dt.day,
            hours, minutes, seconds
        )

        # Day Of Month and Day of Week
        if _hours != '*' and _day_of_month == '*':
            day_of_month = from_dt.day + 1
            day_of_month = 1 if day_of_month == 32 else day_of_month
        elif _hours == '*' and (_day_of_month == '*' or  _day_of_week == '*'):
            day_of_month = from_dt.day
        elif _day_of_month == '?' and '-' in _day_of_week:
            days_of_week = _day_of_week.split('-')
            next_day_of_week = self.adjust_cron_to_pydt(days_of_week[0])
            if next_day_of_week == from_dt.weekday():
                day_of_month = from_dt.day
            else:
                next_week_day_date = from_dt
                while next_week_day_date.weekday() != day_of_week:
                    next_week_day_date += datetime.timedelta(days=1)
                    day_of_month = next_week_day_date.day
        elif _day_of_month == '?' and _day_of_week in \
            ['sun', 'mon', 'tue', 'wed', 'thu', 'fri', 'sat']:
            day_of_week = {
                'mon': 0, 'tue': 1, 'wed': 2, 
                'thu': 3, 'fri': 4, 'sat': 5, 'sun': 6
            }[_day_of_week]
            next_week_day_date = from_dt
            if day_of_week == from_dt.weekday():
                day_of_month = from_dt.day
            else:
                while next_week_day_date.weekday() != day_of_week:
                    next_week_day_date += datetime.timedelta(days=1)
                    day_of_month = next_week_day_date.day
        else:
            day_of_month = int(_day_of_month)
            hours = 0 if from_dt.hour == next_dt.hour else hours
            minutes = 0 if from_dt.minute == next_dt.minute else minutes
            # seconds = 0 if current_date.second == next_date.second else seconds
            seconds = 0
            if not (day_of_month >= 1 and day_of_month <= 31):
                raise Exception('Day of month value must be between 1 and 31!')

        next_dt = datetime.datetime(
            next_dt.year, next_dt.month, day_of_month,
            hours, minutes, seconds
        )

        # Month
        if _day_of_month == '?' and _month == '*':
            month = from_dt.month
        elif  _day_of_month != '*' and _month == '*':
            month = from_dt.month + 1
            month = 1 if month == 13 else month
        elif _day_of_month == '*' and _month == '*':
            month = from_dt.month
        else:
            month = int(_month)
            day_of_month = 1
            hours = 0 if from_dt.hour == next_dt.hour else hours
            minutes = 0 if from_dt.minute == next_dt.minute else minutes
            # seconds = 0 if current_date.second == next_date.second else seconds
            seconds = 0
            if not (month >= 1 and month <= 12):
                raise Exception('Month value must be between 1 and 12!')

        next_dt = datetime.datetime(
            next_dt.year, month, day_of_month,
            hours, minutes, seconds
        )

        # Year
        if  _month != '*' and _year == '*':
            year = from_dt.year + 1
            year = 1 if year == 9999 else year
        elif _month == '*' and _year == '*':
            year = from_dt.year
        else:
            year = int(_year)
            month = 1
            day_of_month = 1
            hours = 0 if from_dt.hour == next_dt.hour else hours
            minutes = 0 if from_dt.minute == next_dt.minute else minutes
            # seconds = 0 if current_date.second == next_date.second else seconds
            seconds = 0
            if not (year >= 1 and year <= 9999):
                raise Exception('Year value must be between 1 and 9999!')

        next_dt = datetime.datetime(
            year, month, day_of_month, 
            hours, minutes, seconds
        )

        return next_dt

    def adjust_cron_to_pydt(self, day: str) -> int:
        new_dw = int(day)-1
        if new_dw == -1:
            new_dw = 6
        return new_dw

if __name__ == '__main__':
    from_date = datetime.datetime(2023, 5, 10, 2, 0, 0)
    cron = Cronzun('20 * * * * ? *', from_date)
    next_date = cron.get_next_date()
    print('from date:', from_date)
    print('next date:', next_date)

E também tem o arquivo de testes test_cronzun.py:

import typing
import datetime
from cronzun import Cronzun

def test1():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 23, 31, 47)

def test2():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('0 * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 23, 32, 0)

def test3():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('1 * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 23, 32, 1)

def test4():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('30 * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 23, 32, 30)

def test5():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('59 * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 23, 32, 59)

def test5():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* 0 * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 0, 0, 47)

def test6():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* 1 * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 0, 1, 47)

def test7():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* 30 * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 0, 30, 47)

def test8():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* 59 * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 28, 0, 59, 47)

def test9():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * 0 * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 29, 0, 31, 47)

def test10():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * 1 * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 29, 1, 31, 47)

def test11():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * 12 * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 29, 12, 31, 47)

def test12():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * 23 * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 9, 29, 23, 31, 47)

def test13():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * 1 * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 1, 0, 0, 0)

def test14():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * 15 * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 15, 0, 0, 0)

def test15():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * 30 * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 30, 0, 0, 0)

def test16():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * 1 ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2024, 1, 1, 0, 0, 0)

def test17():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * 6 ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2024, 6, 1, 0, 0, 0)

def test18():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * 12 ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2024, 12, 1, 0, 0, 0)

def test19():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * * ? 2024', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2024, 1, 1, 0, 0, 0)

def test20():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * * ? 5000', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(5000, 1, 1, 0, 0, 0)

def test21():
    dt = datetime.datetime(2023, 9, 28, 23, 31, 46)
    cron = Cronzun('* * * * * ? 9999', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(9999, 1, 1, 0, 0, 0)

def test22():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 0 * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 2, 10, 0, 0)

def test23():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 0 0 * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 3, 0, 0, 0)

def test24():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 0 0 1 * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 11, 1, 0, 0, 0)

def test25():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 0 0 1 1 ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2024, 1, 1, 0, 0, 0)

def test26():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 0 0 1 1 ? 2025', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2025, 1, 1, 0, 0, 0)

def test27():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 5 0 * 8 ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2024, 8, 1, 0, 5, 0)

def test28():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 15 14 1 * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 11, 1, 14, 15, 0)

def test29():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 0 22 ? * 1-5 *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 2, 22, 0, 0)

def test30():
    dt = datetime.datetime(2023, 10, 2, 9, 30, 30)
    cron = Cronzun('0 5 4 ? * sun *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 8, 4, 5, 0)

def test30():
    dt = datetime.datetime(2023, 10, 2, 12, 2, 20)
    cron = Cronzun('5 * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 2, 12, 3, 5)

def test31():
    dt = datetime.datetime(2023, 10, 2, 12, 2, 0)
    cron = Cronzun('10 * * * * ? *', dt)
    cnd = cron.get_next_date()
    print(cnd)
    assert cnd == datetime.datetime(2023, 10, 2, 12, 2, 10)

Acredito que este desafio - sem compromisso - vale muito a pena, pois é um projeto com muito potencial de melhoria, sem a pretenção de substituir soluções que já agregam bastante no mercado, com a ideia de ser divertido, que possa ensinar outras pessoas assim como me ensinou bastante.

Quero com a apresentação deste pitch e deste desafio motivar você a criar seu próprio projeto de Cron, ou qualquer outro projeto que você fique curioso para entender seu comportamento por trás dos panos e assim você conseguir melhorar suas skills de desenvolvimento!