112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
from databases.no_sql_models.validations import (
|
|
PasswordHistoryViaUser,
|
|
DomainViaUser,
|
|
AccessHistoryViaUser,
|
|
)
|
|
from databases.no_sql_models.mongo_database import MongoQuery
|
|
from api_library.date_time_actions.date_functions import system_arrow
|
|
|
|
|
|
class MongoQueryIdentity:
|
|
"""
|
|
4ex. mongo_collection_name = str(Company.uu_id()) + "*" + str('UserPasswordHistory')
|
|
"""
|
|
|
|
def __init__(self, company_uuid, storage_reasoning: str = None):
|
|
self.company_uuid = company_uuid
|
|
self.mongo_collection_base = str(company_uuid)
|
|
self.mongo_collection_name = (
|
|
str(company_uuid) + "*" + str(storage_reasoning or "Domain")
|
|
)
|
|
self.mongo_engine = MongoQuery(
|
|
table_name=self.mongo_collection_name, database_name="mongo_database"
|
|
)
|
|
|
|
def use_collection(self, storage_reasoning):
|
|
self.mongo_collection_name = (
|
|
str(self.company_uuid) + "*" + str(storage_reasoning)
|
|
)
|
|
self.mongo_engine = MongoQuery(
|
|
table_name=self.mongo_collection_name, database_name="mongo_database"
|
|
)
|
|
|
|
def create_domain_via_user(self, payload: DomainViaUser):
|
|
return self.mongo_engine.find_or_insert(
|
|
field="user_uu_id",
|
|
payload={
|
|
"user_uu_id": payload.user_uu_id,
|
|
"other_domains_list": [payload.main_domain],
|
|
"main_domain": payload.main_domain,
|
|
"modified_at": system_arrow.to_timestamp(system_arrow.now()),
|
|
},
|
|
)
|
|
|
|
def update_domain_via_user(self, payload: DomainViaUser):
|
|
return self.mongo_engine.update(
|
|
match=payload.user_uu_id,
|
|
payload={
|
|
"other_domains_list": payload.other_domains_list,
|
|
"modified_at": system_arrow.to_timestamp(system_arrow.now()),
|
|
},
|
|
field="user_uu_id",
|
|
)
|
|
|
|
def get_domain_via_user(self, user_uu_id):
|
|
result = self.mongo_engine.filter_by(payload={"user_uu_id": user_uu_id})
|
|
return [self.validate_timestamp(doc) for doc in result.data] if result else None
|
|
|
|
def refresh_password_history_via_user(self, payload: PasswordHistoryViaUser):
|
|
return self.mongo_engine.update(
|
|
field="user_uu_id",
|
|
match=payload.user_uu_id,
|
|
payload={
|
|
"password_history": payload.password_history,
|
|
"access_history_detail": payload.access_history_detail,
|
|
"modified_at": system_arrow.to_timestamp(system_arrow.now()),
|
|
},
|
|
)
|
|
|
|
def get_password_history_via_user(self, user_uu_id):
|
|
result = self.mongo_engine.filter_by(payload={"user_uu_id": user_uu_id})
|
|
return [self.validate_timestamp(doc) for doc in result.data] if result else None
|
|
|
|
def update_access_history_via_user(self, payload: AccessHistoryViaUser):
|
|
if already_dict := self.get_access_history_via_user(
|
|
user_uu_id=payload.user_uu_id
|
|
):
|
|
access_history = already_dict[0].get("access_history") or []
|
|
access_history.append(payload.access_history)
|
|
if len(access_history) > 60:
|
|
access_history.pop(0)
|
|
return self.mongo_engine.update(
|
|
match=payload.user_uu_id,
|
|
payload={
|
|
"user_uu_id": payload.user_uu_id,
|
|
"access_history": access_history,
|
|
"modified_at": system_arrow.to_timestamp(system_arrow.now()),
|
|
},
|
|
field="user_uu_id",
|
|
)
|
|
return self.mongo_engine.insert(
|
|
payload={
|
|
"user_uu_id": payload.user_uu_id,
|
|
"access_history": [payload.access_history],
|
|
"modified_at": system_arrow.to_timestamp(system_arrow.now()),
|
|
}
|
|
)
|
|
|
|
def get_access_history_via_user(self, user_uu_id):
|
|
result = self.mongo_engine.filter_by(payload={"user_uu_id": user_uu_id})
|
|
return [self.validate_timestamp(doc) for doc in result.data] if result else None
|
|
|
|
def validate_timestamp(doc):
|
|
"""Validate and fix timestamp fields in MongoDB documents"""
|
|
if not doc:
|
|
return doc
|
|
|
|
timestamp_fields = ["modified_at", "created_at", "accessed_at", "timestamp"]
|
|
for field in timestamp_fields:
|
|
if field in doc and not isinstance(doc[field], (int, float)):
|
|
doc[field] = system_arrow.to_timestamp(doc[field])
|
|
return doc
|