88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
from __future__ import annotations
|
|
import os
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from typing import Final, Optional
|
|
from flask import current_app
|
|
|
|
from appwrite.client import Client
|
|
from appwrite.services.tables_db import TablesDB
|
|
from appwrite.query import Query
|
|
from appwrite.id import ID
|
|
|
|
from app.utils.logging import get_logger
|
|
from app.utils.settings import get_settings, Environment
|
|
|
|
settings = get_settings()
|
|
logger = get_logger(__file__)
|
|
|
|
|
|
class Env(StrEnum):
|
|
PROD = "prod"
|
|
DEV = "dev"
|
|
|
|
# --- Database schemas (strongly-typed namespaces) ----------------------------
|
|
|
|
@dataclass(frozen=True)
|
|
class Database:
|
|
"""Schema for a single database: each attribute is a table name (or ID)."""
|
|
id: str
|
|
characters: str
|
|
inventory: str
|
|
# add more tables here as you grow your schema
|
|
|
|
@dataclass(frozen=True)
|
|
class Databases:
|
|
"""Top-level namespace exposing prod/dev as attributes."""
|
|
prod: Database
|
|
dev: Database
|
|
|
|
DB: Final[Databases] = Databases(
|
|
prod=Database(
|
|
id="SETME", # actual DB / ID
|
|
characters="SETME", # actual table / ID
|
|
inventory="inventory",
|
|
),
|
|
dev=Database(
|
|
id="69041f9600177b675485",
|
|
characters="69050f830024afb0d253",
|
|
inventory="inventory",
|
|
),
|
|
)
|
|
|
|
class AppwriteTables:
|
|
|
|
def __init__(self):
|
|
print()
|
|
self.client = (Client()
|
|
.set_endpoint(settings.appwrite_endpoint)
|
|
.set_project(settings.appwrite_project_id)
|
|
.set_key(settings.appwrite_api_key)
|
|
)
|
|
self.tables_db = TablesDB(self.client)
|
|
self.env = Env.DEV
|
|
if settings.env == Environment.PROD:
|
|
self.env = Env.PROD
|
|
|
|
@property
|
|
def db(self) -> Database:
|
|
# Gives autocompletion for .character, .users, etc.
|
|
return DB.prod if self.env is Env.PROD else DB.dev
|
|
|
|
def get_characters_for_user_id(self, user_id: str) -> Optional[dict]:
|
|
|
|
try:
|
|
result = self.tables_db.list_rows(
|
|
self.db.id,
|
|
self.db.characters,
|
|
[
|
|
Query.equal('player_id', [str(user_id)]),
|
|
]
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Unable to list rows for char. User id: {user_id}")
|
|
return {}
|
|
|
|
return result.get("rows",{})
|
|
|