|
| 1 | +from __future__ import annotations |
| 2 | +from abc import ABC |
| 3 | +from typing import get_type_hints, get_origin, get_args, TypeVar |
| 4 | +from types import UnionType, NoneType, |
| 5 | +from . import io_safe |
| 6 | + |
| 7 | +T = TypeVar('T') |
| 8 | + |
| 9 | + |
| 10 | + |
| 11 | +class FolderBase(ABC): |
| 12 | + __folder__ = None |
| 13 | + __file__base__ = None |
| 14 | + |
| 15 | + |
| 16 | +class FileDictBase(ABC): |
| 17 | + """ |
| 18 | + A file base refers to a file that is stored in the database. |
| 19 | + """ |
| 20 | + |
| 21 | + __file__ = None |
| 22 | + __item_base__: T = None |
| 23 | + __create_file_on_write__ = False |
| 24 | + |
| 25 | + |
| 26 | + @classmethod |
| 27 | + def get_by_key(cls, key) -> T: |
| 28 | + """ |
| 29 | + Gets an item by key. |
| 30 | + """ |
| 31 | + return cls.__item_base__.get_by_key(key) |
| 32 | + |
| 33 | + |
| 34 | + |
| 35 | + |
| 36 | + |
| 37 | +class FileDictItemBase(ABC): |
| 38 | + @classmethod |
| 39 | + def get_by_key(cls, key): |
| 40 | + data = io_safe.partial_read(cls.__file__, key) |
| 41 | + return cls.from_data(data) |
| 42 | + |
| 43 | + |
| 44 | + |
| 45 | + @classmethod |
| 46 | + def from_data(cls, data): |
| 47 | + |
| 48 | + instance = cls() |
| 49 | + |
| 50 | + for var_name, var_type in get_type_hints(cls).items(): |
| 51 | + # Check if variable is nullable (e.g. email: str | None) |
| 52 | + nullable = get_origin(var_type) is UnionType and NoneType in get_args(var_name) |
| 53 | + # When it is not nullable but not in the data, raise an error |
| 54 | + if var_name not in data and not nullable: |
| 55 | + raise RuntimeError(f"Missing variable '{var_name}' in file '{cls.__file__}'.") |
| 56 | + |
| 57 | + setattr(instance, var_name, data.get(var_name, None)) |
| 58 | + |
| 59 | + return instance |
| 60 | + |
| 61 | + |
| 62 | + |
| 63 | + |
| 64 | + |
| 65 | + |
| 66 | +class User(FileDictItemBase): |
| 67 | + first_name: str | None |
| 68 | + last_name: str |
| 69 | + email: str |
| 70 | + |
| 71 | + def full_name(self): |
| 72 | + return f"{self.first_name} {self.last_name}" |
| 73 | + |
| 74 | + |
| 75 | +class Users(FileDictBase): |
| 76 | + __file__ = "users" |
| 77 | + __item_base__ = User |
| 78 | + |
| 79 | + |
| 80 | + |
| 81 | +# Iterate FileDictBase |
| 82 | +# for user_id, user: User in Users.items(): |
| 83 | +# print(user_id, user.first_name, user.last_name, user.email) |
| 84 | + |
| 85 | +# # Get one item |
| 86 | +# user: User = Users.get("user_id") |
| 87 | + |
| 88 | + |
| 89 | +# Get by lambda |
| 90 | +# users: Users = Users.where(lambda user: user.first_name != "John") |
0 commit comments