Example of Programming By Contract (契約プログラミングサンプル)

contract.py

from logging import getLogger

import pandas as pd

from core.db import get_new_user_id

from user_management import tasks  # celery tasks

logger = getLogger(__name__)


def save_new_users(user_csv_path, encoding='utf8'):
    df_user_names = pd.read_csv(user_csv_path, encoding=encoding)
    users = df_user_names.apply(lambda row: User(**row), axis=1).values.tolist())

    valid_users = []
    for user in users:
        # precondition
        try:
            user.validate()
        except InvalidUser as e:
            logger.info(f'[update_users] {e}')
            continue
        user.id = get_new_user_id()
        valid_users.append(user)

    User.objects.bluk_create(valid_users)
    saved_users = User.objects.filter(id__in=[u.id for u in valid_users])

    for user in saved_users:
        tasks.generate_user_avatar(user)
        # Avatar generate by user_name and user_name is already validated,
        # so that tasks.generate_user_avatar() doesn't need to check length of user_name or encoding of user_name, etc...