Skip to content

Fixtures

Fixtures - To populate the database with pre-created data.

Runs automatically during Model migration.

apply_fixture(fixture_name, cls_model, collection) async

Apply fixture for current Model.

Runs automatically during Model migration.

Source code in src/ramifice/fixtures.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def apply_fixture(
    fixture_name: str,
    cls_model: Any,
    collection: AsyncCollection,
) -> None:
    """Apply fixture for current Model.

    Runs automatically during Model migration.
    """
    metadata = cls_model.META
    fixture_path: str = f"config/fixtures/{fixture_name}.yml"
    data_yaml: dict[str, Any] | list[dict[str, Any]] | None = None

    with Path.open(Path(fixture_path)) as file:
        data_yaml = yaml.safe_load(file)

    if not bool(data_yaml):
        err_msg = (
            f"Model: `{metadata['full_model_name']}` > "
            + f"META param: `fixture_name` ({fixture_name}) => "
            + "It seems that fixture is empty or it has incorrect contents!"
        )
        logger.critical(err_msg)
        raise PanicError(err_msg)

    if data_yaml is not None:
        if not isinstance(data_yaml, list):
            data_yaml = [data_yaml]

        for data in data_yaml:
            instance_model = cls_model()
            for f_name in metadata["all_descriptor_fields"]:
                f__core = getattr(instance_model, f"{f_name}__core")
                f__core = getattr(instance_model, f"{f_name}__core")
                group = f__core.group
                value: Any | None = data.get(f__core.name)
                if value == "None":
                    value = None
                if value is not None:
                    if group == "file" or group == "img":
                        await f__core.from_path(value)
                        value = f__core.value
                    setattr(instance_model, f_name, value)
            # Check Model.
            result_check: dict[str, Any] = await instance_model.check(
                is_save=True,
                collection=collection,
            )
            # If the check fails.
            if not result_check["is_valid"]:
                await collection.database.drop_collection(collection.name)
                print(colored("\nFIXTURE:", "red", attrs=["bold"]))  # ruff:ignore[print]
                print(colored(fixture_path, "blue", attrs=["bold"]))  # ruff:ignore[print]
                instance_model.print_err()
                err_msg = f"Fixture `{fixture_name}` failed."
                logger.critical(err_msg)
                raise PanicError(err_msg)
            # Get data for document.
            checked_data: dict[str, Any] = result_check["data"]
            # Add date and time.
            today = datetime.now(instance_model.utc_timezone)
            checked_data["created_at"] = today
            checked_data["updated_at"] = today
            # Run hook.
            await instance_model.pre_create()
            # Insert doc.
            try:
                await collection.insert_one(checked_data)
            except:
                await collection.database.drop_collection(collection.name)
            # Run hook.
            await instance_model.post_create()