(20241212) dictionary key bug fix.
This commit is contained in:
@@ -0,0 +1,266 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Thursday, 12th Dec., 2024.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To define how information about files will be stored in the database.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# For making data behaviour_models:
|
||||
from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime
|
||||
from typing import Optional, Literal, Union, Any, Dict, List
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import regex
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# To work with MongoDB:
|
||||
from bson.objectid import ObjectId
|
||||
|
||||
# Other data models:
|
||||
from models.data.core.user import CoreUserInfoModel
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class CoreFileInfoModel(BaseModel):
|
||||
|
||||
fileId: ObjectId = Field(
|
||||
description = "the id of the document in mongodb that holds this information",
|
||||
frozen = False,
|
||||
default = None,
|
||||
alias = "_id"
|
||||
)
|
||||
|
||||
user: CoreUserInfoModel = Field(
|
||||
description = "to identify the user who owns this file",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
filename: str = Field(
|
||||
description = "the name of this file",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
uploadTs: AwareDatetime = Field(
|
||||
description = "the time (utc) at which this file was uploaded",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
length: int | None = Field(
|
||||
description = "to note the size of the file in bytes",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
hash: str | None = Field(
|
||||
description = "a simple hash to verify the integrity of the uploaded data; irrelevant for dirs",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
metadata: Dict[str, Any] = Field(
|
||||
description = "any additional data about this file to filter it later",
|
||||
frozen = False,
|
||||
default = {}
|
||||
)
|
||||
|
||||
tags: List[str] = Field(
|
||||
description = "a list of keywords to apply to this file to filter it later",
|
||||
frozen = False,
|
||||
default = [],
|
||||
examples = ["Bank Statement", "PDF", "bhopli@orange.com"]
|
||||
)
|
||||
|
||||
isPrivate: bool = Field(
|
||||
description = "whether this file is a private file, or publicly available",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "allow"
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
def model_dump(self, *args, **kwargs):
|
||||
return super().model_dump(*args, by_alias = True, **kwargs)
|
||||
|
||||
# ┓┏ ┓• ┓ •
|
||||
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
|
||||
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
|
||||
|
||||
@field_validator("uploadTs", mode = "before")
|
||||
def parse_date_time(cls, value):
|
||||
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
|
||||
|
||||
@field_validator("metadata", mode = "before")
|
||||
def validate_metadata(cls, value):
|
||||
if value is None: value = {}
|
||||
return value
|
||||
|
||||
@field_validator("tags", mode = "before")
|
||||
def validate_tags(cls, value):
|
||||
if value is None: value = []
|
||||
return value
|
||||
|
||||
@field_validator("fileId", mode = "before")
|
||||
def parse_oid(cls, value):
|
||||
try:
|
||||
if isinstance(value, str):
|
||||
value = ObjectId(value)
|
||||
except: pass
|
||||
return value
|
||||
|
||||
# ┏┓ •
|
||||
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
|
||||
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
|
||||
# ┛
|
||||
|
||||
@property
|
||||
def summary(self):
|
||||
return {
|
||||
"user": self.user.model_dump(),
|
||||
"name": self.name,
|
||||
"createTs": self.createTs,
|
||||
"parentId": self.parentId,
|
||||
"isPrivate": self.isPrivate,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class CoreFileAccessResponseModel(BaseModel):
|
||||
|
||||
success: bool = Field(
|
||||
description = "whether, or not, the operation was successful",
|
||||
frozen = False,
|
||||
default = False
|
||||
)
|
||||
|
||||
message: str | None = Field(
|
||||
description = "useful to note the reason in case of an unsuccessful operation",
|
||||
frozen = False,
|
||||
default = "message not documented"
|
||||
)
|
||||
|
||||
data: Any = Field(
|
||||
description = "the result of the operation",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
exception: Exception | None = Field(
|
||||
description = "to document any exceptions that came up",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "allow"
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
from utils_v2.string import json
|
||||
|
||||
file_info = CoreFileInfoModel(
|
||||
_id = "67519cf3a7804fcbc6f12452",
|
||||
user = CoreUserInfoModel(
|
||||
fullName = "Bhopli Narangi",
|
||||
userId = 1,
|
||||
entityId = 2,
|
||||
billingAccountId = 3,
|
||||
departmentId = 4,
|
||||
branchId = 5,
|
||||
industry = "technology"
|
||||
),
|
||||
filename = "Graduation Certificate.png",
|
||||
uploadTs = date_time.get_current_utc_date_time(as_string = False),
|
||||
length = 1024,
|
||||
hash = "abcdefgh12345678",
|
||||
metadata = {
|
||||
"camera": "iPhone 1000 Pro Max XS"
|
||||
},
|
||||
tags = [
|
||||
"important"
|
||||
],
|
||||
parentId = None,
|
||||
isPrivate = True
|
||||
)
|
||||
|
||||
print("FILE INFO MODEL:", json.to_string(file_info.model_dump(), default = str))
|
||||
@@ -291,6 +291,62 @@ class CoreFileObjectInfoModel(BaseModel):
|
||||
"isPrivate": self.isPrivate,
|
||||
}
|
||||
|
||||
# ┏┓ ┏┓
|
||||
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
|
||||
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
|
||||
|
||||
def is_owner(
|
||||
self,
|
||||
user: CoreUserInfoModel,
|
||||
ignore_null: bool = False
|
||||
):
|
||||
|
||||
"""
|
||||
To check if a given user is the owner of this file object.
|
||||
:param user: The instance of the model that defines the user that needs to be checked.
|
||||
:param ignore_null: Whether, or not, to consider null values in the user to check.
|
||||
:return: True if the given user is the owner, else False.
|
||||
"""
|
||||
|
||||
if self.user.matches(user, ignore_null = ignore_null): return True
|
||||
else: return False
|
||||
|
||||
def has_permission(
|
||||
self,
|
||||
user: CoreUserInfoModel,
|
||||
permission: Literal["read", "write", "delete", "changePermissions"],
|
||||
ignore_null: bool = False
|
||||
) -> bool:
|
||||
|
||||
"""
|
||||
To check if a given user is the owner of this file object.
|
||||
:param user: The instance of the model that defines the user that needs to be checked.
|
||||
:param permission: The permission to check.
|
||||
:param ignore_null: Whether, or not, to consider null values in the user to check.
|
||||
:return: True if the given user is the owner, else False.
|
||||
"""
|
||||
|
||||
# The owner of the file object always has all permissions:
|
||||
if self.is_owner(user, ignore_null = ignore_null): return True
|
||||
|
||||
# Now we check if the given non-owner user has the requested permission:
|
||||
for sharing_data in self.sharedWith:
|
||||
|
||||
# We match the users.
|
||||
# If they don't match, we move to the next user:
|
||||
if not sharing_data.user.matches(
|
||||
user,
|
||||
ignore_null = ignore_null
|
||||
): continue
|
||||
|
||||
# If we found a matching user,
|
||||
# we check for the permission:
|
||||
if getattr(sharing_data.permissions, permission, False):
|
||||
return True
|
||||
|
||||
# If not condition hit, we don't have the permission:
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
@@ -376,7 +432,7 @@ if __name__ == "__main__":
|
||||
billingAccountId = 8,
|
||||
departmentId = 9,
|
||||
branchId = 10,
|
||||
industry = "entertainment"
|
||||
industry = "finance"
|
||||
),
|
||||
permissionType = "explicit",
|
||||
inheritedFromId = None,
|
||||
@@ -391,3 +447,41 @@ if __name__ == "__main__":
|
||||
)
|
||||
|
||||
print("FILE INFO MODEL:", json.to_string(file_info.model_dump(), default = str))
|
||||
|
||||
user_bhopli = CoreUserInfoModel(
|
||||
fullName = "Bhopli Narangi",
|
||||
userId = 1,
|
||||
entityId = 2,
|
||||
billingAccountId = 3,
|
||||
departmentId = 4,
|
||||
branchId = 5,
|
||||
industry = "technology"
|
||||
)
|
||||
|
||||
user_bhopli_partial = CoreUserInfoModel(
|
||||
fullName = "Bhopli Narangi",
|
||||
userId = 1,
|
||||
entityId = None,
|
||||
billingAccountId = None,
|
||||
departmentId = 4,
|
||||
branchId = None,
|
||||
industry = "technology"
|
||||
)
|
||||
|
||||
user_polki = CoreUserInfoModel(
|
||||
fullName = "Polki Muchhwaali",
|
||||
userId = 6,
|
||||
entityId = 7,
|
||||
billingAccountId = 8,
|
||||
departmentId = 9,
|
||||
branchId = 10,
|
||||
industry = "finance"
|
||||
)
|
||||
|
||||
print("IS OWNER:", file_info.is_owner(user_bhopli))
|
||||
print("IS OWNER:", file_info.is_owner(user_bhopli_partial, ignore_null = False))
|
||||
print("IS OWNER:", file_info.is_owner(user_bhopli_partial, ignore_null = True))
|
||||
print("HAS PERM:", file_info.has_permission(user_bhopli_partial, permission = "read", ignore_null = False))
|
||||
print("HAS PERM:", file_info.has_permission(user_bhopli_partial, permission = "read", ignore_null = True))
|
||||
print("HAS PERM:", file_info.has_permission(user_polki, permission = "read", ignore_null = True))
|
||||
print("HAS PERM:", file_info.has_permission(user_polki, permission = "delete", ignore_null = True))
|
||||
|
||||
@@ -131,6 +131,46 @@ class CoreUserInfoModel(BaseModel):
|
||||
extra = "ignore"
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
# ┏┓ ┏┓
|
||||
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
|
||||
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
|
||||
|
||||
def matches(
|
||||
self,
|
||||
other_user: 'CoreUserInfoModel',
|
||||
ignore_null: bool = False
|
||||
) -> bool:
|
||||
|
||||
"""
|
||||
To check if another instance of this user model matches this instance of the user model.
|
||||
:param other_user: The other user to check. That user's value must match this user's values to return a positive
|
||||
result. Whether, or not, null values are matched will be determined by the next param.
|
||||
:param ignore_null: If set to True, null values IN THE OTHER USER will not be compared during the matching.
|
||||
:return: True if they match, else False.
|
||||
"""
|
||||
|
||||
# Start by assuming success:
|
||||
are_matching = True
|
||||
|
||||
# Iterate through the required items:
|
||||
for k, v in other_user.model_dump().items():
|
||||
|
||||
# Do not consider fields that are nulls if asked to ignore them:
|
||||
if ignore_null and v is None: continue
|
||||
|
||||
# Extract the corresponding value from the other user,
|
||||
# and test it for being equal:
|
||||
_v = getattr(self, k, None)
|
||||
if (
|
||||
(not isinstance(v, type(_v))) or
|
||||
(v != _v)
|
||||
):
|
||||
are_matching = False
|
||||
break
|
||||
|
||||
# Done here:
|
||||
return are_matching
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
@@ -143,14 +183,37 @@ if __name__ == "__main__":
|
||||
|
||||
from utils_v2.string import json
|
||||
|
||||
user_info = CoreUserInfoModel(
|
||||
user_bhopli = CoreUserInfoModel(
|
||||
fullName = "Bhopli Narangi",
|
||||
userId = 1,
|
||||
entityId = 2,
|
||||
billingAccountId = 3,
|
||||
departmentId = 4,
|
||||
branchId = 5,
|
||||
industry = "fashion"
|
||||
industry = "technology"
|
||||
)
|
||||
|
||||
print("USER INFO MODEL:", json.to_string(user_info.model_dump(), default = str))
|
||||
user_bhopli_partial = CoreUserInfoModel(
|
||||
fullName = "Bhopli Narangi",
|
||||
userId = 1,
|
||||
entityId = None,
|
||||
billingAccountId = None,
|
||||
departmentId = 4,
|
||||
branchId = None,
|
||||
industry = "technology"
|
||||
)
|
||||
|
||||
user_polki = CoreUserInfoModel(
|
||||
fullName = "Polki Muchhwaali",
|
||||
userId = 6,
|
||||
entityId = 7,
|
||||
billingAccountId = 8,
|
||||
departmentId = 9,
|
||||
branchId = 10,
|
||||
industry = "finance"
|
||||
)
|
||||
|
||||
print(user_bhopli.matches(user_bhopli_partial, ignore_null = False))
|
||||
print(user_bhopli.matches(user_bhopli_partial, ignore_null = True))
|
||||
print(user_bhopli.matches(user_polki, ignore_null = False))
|
||||
print(user_bhopli.matches(user_polki, ignore_null = True))
|
||||
|
||||
Reference in New Issue
Block a user