Introduction

Building projects is important for a ton of reasons. One of them is to gain or fortify knowledge. Here are some of the key features of the finance tracking application (called FinTracker) that I built in my quest for knowledge:

• Authentication and Authorization

• Creation of Income and Expense records

• Scheduling of Income and Expense records

• Customizable Income and Expense categories

• Data visualization

In the next few sections, I explain the things I learned through building FinTracker.

Rendering charts with ChartJS

ChartJS is a free, open-source JavaScript library for data visualization. This was my first time ever using ChartJS, and I found it quite easy to use.

Charts are used for visualizing data. To prepare the data that needed to be visualized, I set up an endpoint to query the database for all transactions (incomes and expenses) from the last six days and return the processed results.

def get_last_six_transactions(request):

    end_date = datetime.now()
    start_date = end_date - timedelta(6)

    # Get the income items from the last six days
    incomes = IncomeItem.objects.filter(
        user=request.user,
        created__range=[start_date, end_date]
    ).values('created__date').annotate(total_income=Sum('amount'))

    # Get the expense items from the last six days
    expenses = ExpenseItem.objects.filter(
        user=request.user,
        created__range=[start_date, end_date]
    ).values('created__date').annotate(total_expense=Sum('amount'))

    # Store the last six dates into a list 
    date_range = [end_date - timedelta(i) for i in range(6)]
    labels = [date.strftime("%d-%m-%Y") for date in date_range]

    # Create a dictionary to store {date: expense_total} for last 6 days
    expense_totals = {expense['created__date'].strftime("%d-%m-%Y"): expense['total_expense'] for expense in expenses}
    expense_amounts = [expense_totals.get(date, 0) for date in labels]

    # Create a dictionary to store {date: income_total} for last 6 days
    income_totals = {income['created__date'].strftime("%d-%m-%Y"): income['total_income'] for income in incomes}
    income_amounts = [income_totals.get(date, 0) for date in labels]

    return JsonResponse({
        'labels': labels,
        'income_amounts': income_amounts,
        'expense_amounts': expense_amounts
    })

Then, I created a ChartJS instance using the response data from the API call.

fetch("{% url  'get_last_six_transactions' %}")
    .then(response => response.json())
    .then((data) => {
        const labels = data['labels'];
        const expenses = data['expense_amounts'];
        const incomes = data['income_amounts'];
        let expenseIncomeChart = new Chart(document.getElementById('expenseIncomeChart'), {
            type: 'bar',
            data: {
                labels: labels,
                datasets: [{
                    label: 'Expense',
                    backgroundColor: 'red',
                    data: expenses
                }, {
                    label: 'Income',
                    backgroundColor: 'green',
                    data: incomes
                }]
            },
            options: {
                responsive: true
            }
        });
    })

Here's how the FinTracker dashboard looks for a user of the application.

Scheduling periodic tasks with Celery

I wanted a feature where users could set up automated transactions. Rather than manually creating records for expenses or incomes that are recurring, users should be granted the ability to tell the application to do it for them. To implement this feature, I used Celery (celery-beat) and Redis.

Celery is a task queue implementation for Python web applications used to asynchronously execute work outside the HTTP request-response cycle. In less nerdy terms, this means that instead of waiting for a task to complete before responding to a user's request, Celery lets the application handle it later, while still responding quickly to the user- but that's not all.

Celery possesses an inbuilt mechanism called celery-beat which is responsible for scheduling and managing the execution of periodic tasks. These are tasks that need to run at fixed intervals (e.g., every hour, every day) or tasks scheduled for specific times (e.g., 3pm on Wednesdays).

To install Celery with the dependencies required for using Redis as a message transport or as a result backend:

$ pip install celery[redis]

See this for clear information on how to set up Celery and Redis for your Django project.

The plan was to allow users to either create transaction records manually, or set up recurring transactions. For that to work, this was how I set up my expense models (Note: The income models were exact replicas of these):

# For regular expense items
class ExpenseItem(TimeStampedModel):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    title = models.CharField(max_length=120)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    category = models.ForeignKey("categories.Category", on_delete=models.SET_DEFAULT, default=None)
    notes = models.CharField(max_length=255, blank=True, null=True)

    class Meta:
        ordering = ['-created']

    def __str__(self):
        return f"{self.user.first_name}: {self.title} Cost: {self.amount}"

# For recurring expense items
class ScheduledExpenseItem(models.Model):

    class Frequency(models.TextChoices):
        DAILY = 'daily', "Daily"
        WEEKLY = 'weekly', "Weekly"
        MONTHLY = 'monthly', "Monthly"
        YEARLY = 'yearly', "Yearly"

    source_expense_item = models.ForeignKey(ExpenseItem, on_delete=models.CASCADE)
    frequency = models.CharField(max_length=20, null=True, choices=Frequency.choices)
    created = models.DateTimeField(auto_now_add=True)
    last_process_time = models.DateTimeField(auto_now_add=True)
    is_active = models.BooleanField(default=False)

    def __str__(self):
        return f"Automating: {self.source_expense_item}"

Here's what it looks like when a user schedules a transaction:

The field values are sent to the view, and an initial ExpenseItem, as well as a corresponding ScheduledExpenseItem, is created.

def create_automated_expense(request):
    expense_title = request.POST.get('title')
    expense_amount = request.POST.get('amount')
    expense_category = request.POST.get('category')
    expense_notes = request.POST.get('notes')
    frequency = request.POST.get('frequency')

    new_expense_item = ExpenseItem.objects.create(
        user=request.user,
        title=expense_title,
        amount=expense_amount,
        category=Category.objects.get(
            user=request.user,
            title=expense_category,
            type='E'
            ),
        notes=expense_notes
    )

    ScheduledExpenseItem.objects.create(
        source_expense_item=new_expense_item,
        frequency=frequency,
        is_active=True
    )

    serialized_data = ExpenseItemSerializer(new_expense_item).data

    return JsonResponse({
        'message': 'success',
        'expense': {
            'data': serialized_data,
            'frequency': frequency
        }
    })

That's about it for the regular Django stuff. Next, we take a look at how the scheduling actually works.

Tasks in Celery are defined as regular Python functions or methods that are decorated with the @task or @shared_task decorator provided by Celery. These tasks can then be called like regular functions, but instead of being executed immediately, they are placed into a message queue where Celery workers pick them up and execute them asynchronously.

I set up a couple of tasks that inspect all active ScheduledIncomeItems and ScheduledExpenseItems, and determines if they are due to be re-recorded.

@shared_task()
def handle_scheduled_transactions(*args, **kwargs):

    scheduled_expense_items = ScheduledExpenseItem.objects.all()

    for scheduled_expense_item in scheduled_expense_items:
        if scheduled_expense_item.is_active:
            timegap = timezone.now() - scheduled_expense_item.last_process_time
            source_expense_item = scheduled_expense_item.source_expense_item

            if scheduled_expense_item.frequency == 'daily':
                difference_in_hours = timegap.total_seconds() // 3600
                if difference_in_hours >= 24:
                    create_expense_item(source_expense_item)
                    scheduled_expense_item.last_process_time = datetime.now()
                    scheduled_expense_item.save()

            # ... (Code omitted for brevity)

            elif scheduled_expense_item.frequency == 'yearly':
                difference_in_days = timegap.days
                if difference_in_days >= 365:
                    create_expense_item(source_expense_item)
                    scheduled_expense_item.last_process_time = datetime.now()
                    scheduled_expense_item.save()

Finally, I had to tell Celery (celery-beat) how frequently these tasks should be executed. I went for 'every 30 seconds'. Again, what this means is that Celery will execute these functions- that check if there are expenses or incomes at that particular moment that are due to be re-recorded- every 30 seconds.

app.conf.beat_schedule = {
    'handle-scheduled-expenses': {
        'task': 'expenses.tasks.handle_scheduled_transactions',
        'schedule': 30.0,
    },
    'handle-scheduled-incomes': {
        'task': 'incomes.tasks.handle_scheduled_transactions',
        'schedule': 30.0,
    }
}

Conclusion

If you ask any experienced software developer about how you could become better at software development, there's a 99% chance that they'd advise you to build more projects. 99% of those times, they'd be right.

ChartJS and Celery are technologies that I had heard of before but never actually used until I had a reason to. When you attempt to build projects, you will inevitably come across features you want to implement, or problems you have to solve- but you don't know how. This should prompt you to ask questions, make research and experiment with concepts and technologies you may or may not have heard about before.