logan/drb-core-server
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix active/all token issues
All checks were successful
release-image / release-image (push) Successful in 3m15s
Details

Browse Source
This commit is contained in:
Logan Cusano
2025-07-06 20:16:56 -04:00
parent 1658ea2e83
commit 776b3d9ac2
1 changed files with 12 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app/internal/db_wrappers.py
View File

@@ -74,7 +74,7 @@ class SystemDbController():
if found_docs: if found_docs:
print("Found document (raw dict):", found_docs) print("Found document (raw dict):", found_docs)
converted_systems = [System.from_dict(doc) for doc in found_docs] converted_systems = [System.from_dict(doc) for doc in found_docs]
print("YURB", found_docs, converted_systems) print("Found document (converted):", converted_systems)
return converted_systems if len(converted_systems) > 0 else None return converted_systems if len(converted_systems) > 0 else None
else: else:
print("Document not found.") print("Document not found.")
@@ -83,12 +83,17 @@ class SystemDbController():
print(f"Find failed: {e}") print(f"Find failed: {e}")
return None return None
async def find_all_systems(self, query: Dict[str, Any] = {}) -> List[System]: async def find_all_systems(self, query: Optional[Dict[str, Any]] = None) -> List[System]:
print("\n--- Finding multiple documents ---") print("\n--- Finding multiple documents ---")
try: try:
# Initialize an empty dictionary if no query is provided
if query is None:
query = {}
found_docs = None found_docs = None
async with self.db_h as db: # async with self.db_h as db: #
found_docs = await db.find(query) # found_docs = await db.find(query) #
if found_docs: if found_docs:
print(f"Found {len(found_docs)} documents (raw dicts).") print(f"Found {len(found_docs)} documents (raw dicts).")
return [System.from_dict(doc) for doc in found_docs] return [System.from_dict(doc) for doc in found_docs]
@@ -160,11 +165,9 @@ class DiscordIdDbController():
print(f"Discord ID create failed: {e}") print(f"Discord ID create failed: {e}")
return None return None
async def find_discord_id(self, query: Dict[str, Any], active_only: bool = False) -> Optional[DiscordId]: async def find_discord_id(self, query: Dict[str, Any]) -> Optional[DiscordId]:
print("\n--- Finding one Discord ID document ---") print("\n--- Finding one Discord ID document ---")
try: try:
if active_only:
query["active"] = True
found_doc = None found_doc = None
async with self.db_h as db: # async with self.db_h as db: #
found_doc = await db.find_one(query) # found_doc = await db.find_one(query) #
@@ -178,9 +181,12 @@ class DiscordIdDbController():
print(f"Discord ID find failed: {e}") print(f"Discord ID find failed: {e}")
return None return None
async def find_discord_ids(self, query: Dict[str, Any] = {}, guild_id: Optional[str] = None, active_only: bool = False) -> Optional[List[DiscordId]]: async def find_discord_ids(self, query: Optional[Dict[str, Any]] = None, guild_id: Optional[str] = None, active_only: bool = False) -> Optional[List[DiscordId]]:
print("\n--- Finding multiple Discord ID documents ---") print("\n--- Finding multiple Discord ID documents ---")
try: try:
if query is None:
query = {}
if active_only == True: if active_only == True:
print("Searching active IDs") print("Searching active IDs")
query["active"] = True query["active"] = True