fix

python_core/database/db.py

@@ -1,10 +1,518 @@
 
+import json
+import uuid
+from typing import Dict, List, Any, Optional, Union
+from datetime import datetime
 from python_core.kv import kv
+from python_core.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
 
 class Db:
-    
+    """
+    基于KV存储的数据库类
+    提供基础的CRUD操作、索引、查询等功能
+    """
+
     def __init__(self, key: str):
+        """
+        初始化数据库
+
+        Args:
+            key: 数据库的唯一标识符，用作KV存储的前缀
+        """
         self.kv = kv
         self.key = key
+        self.table_prefix = f"db:{key}:table:"
+        self.index_prefix = f"db:{key}:index:"
+        self.meta_prefix = f"db:{key}:meta:"
+
+        # 初始化数据库元数据
+        self._init_metadata()
+
+    def _init_metadata(self):
+        """初始化数据库元数据"""
+        try:
+            meta_key = f"{self.meta_prefix}info"
+            existing_meta = self.kv.get(meta_key)
+
+            if not existing_meta:
+                # 创建新的数据库元数据
+                metadata = {
+                    "created_at": datetime.now().isoformat(),
+                    "version": "1.0.0",
+                    "tables": [],
+                    "indexes": {}
+                }
+                self.kv.set(meta_key, json.dumps(metadata))
+                logger.info(f"Initialized new database: {self.key}")
+            else:
+                logger.info(f"Connected to existing database: {self.key}")
+
+        except Exception as e:
+            logger.error(f"Failed to initialize database metadata: {e}")
+            raise e
+
+    def _get_metadata(self) -> Dict[str, Any]:
+        """获取数据库元数据"""
+        try:
+            meta_key = f"{self.meta_prefix}info"
+            logger.debug(f"Getting metadata with key: {meta_key}")
+            meta_str = self.kv.get(meta_key)
+            logger.debug(f"Raw metadata string: {meta_str}")
+            if meta_str:
+                parsed = json.loads(meta_str)
+                logger.debug(f"Parsed metadata: {parsed}")
+                return parsed
+            logger.debug("No metadata found, returning empty dict")
+            return {}
+        except Exception as e:
+            logger.error(f"Failed to get database metadata: {e}")
+            import traceback
+            traceback.print_exc()
+            return {}
+
+    def _update_metadata(self, metadata: Dict[str, Any]):
+        """更新数据库元数据"""
+        try:
+            meta_key = f"{self.meta_prefix}info"
+            logger.debug(f"Updating metadata with key: {meta_key}")
+            logger.debug(f"Metadata content: {metadata}")
+            result = self.kv.set(meta_key, json.dumps(metadata))
+            logger.debug(f"Update result: {result}")
+
+            # 添加短暂延迟确保KV同步
+            import time
+            time.sleep(0.1)
+
+        except Exception as e:
+            logger.error(f"Failed to update database metadata: {e}")
+            raise e
+
+    def create_table(self, table_name: str, schema: Dict[str, str] = None) -> bool:
+        """
+        创建表
+
+        Args:
+            table_name: 表名
+            schema: 表结构定义 (字段名: 类型)
+
+        Returns:
+            创建成功返回True
+        """
+        try:
+            metadata = self._get_metadata()
+
+            if table_name in metadata.get("tables", []):
+                logger.warning(f"Table '{table_name}' already exists")
+                return False
+
+            # 创建表元数据
+            table_meta = {
+                "name": table_name,
+                "created_at": datetime.now().isoformat(),
+                "schema": schema or {},
+                "record_count": 0,
+                "indexes": []
+            }
+
+            # 保存表元数据
+            table_meta_key = f"{self.meta_prefix}table:{table_name}"
+            self.kv.set(table_meta_key, json.dumps(table_meta))
+
+            # 更新数据库元数据
+            metadata.setdefault("tables", []).append(table_name)
+            self._update_metadata(metadata)
+
+            logger.info(f"Created table: {table_name}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to create table '{table_name}': {e}")
+            raise e
+
+    def drop_table(self, table_name: str) -> bool:
+        """
+        删除表
+
+        Args:
+            table_name: 表名
+
+        Returns:
+            删除成功返回True
+        """
+        try:
+            metadata = self._get_metadata()
+
+            if table_name not in metadata.get("tables", []):
+                logger.warning(f"Table '{table_name}' does not exist")
+                return False
+
+            # 获取表中所有记录的键
+            record_keys = self._get_all_record_keys(table_name)
+
+            # 删除所有记录
+            if record_keys:
+                self.kv.removes(record_keys)
+
+            # 删除表元数据
+            table_meta_key = f"{self.meta_prefix}table:{table_name}"
+            self.kv.remove(table_meta_key)
+
+            # 删除相关索引
+            self._drop_table_indexes(table_name)
+
+            # 更新数据库元数据
+            metadata["tables"] = [t for t in metadata.get("tables", []) if t != table_name]
+            self._update_metadata(metadata)
+
+            logger.info(f"Dropped table: {table_name}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to drop table '{table_name}': {e}")
+            raise e
+
+    def insert(self, table_name: str, data: Dict[str, Any], record_id: str = None) -> str:
+        """
+        插入记录
+
+        Args:
+            table_name: 表名
+            data: 记录数据
+            record_id: 记录ID，如果不提供则自动生成
+
+        Returns:
+            记录ID
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            # 生成记录ID
+            if not record_id:
+                import uuid
+                record_id = str(uuid.uuid4())
+
+            # 添加元数据
+            record = {
+                "id": record_id,
+                "data": data,
+                "created_at": datetime.now().isoformat(),
+                "updated_at": datetime.now().isoformat()
+            }
+
+            # 保存记录
+            record_key = f"{self.table_prefix}{table_name}:record:{record_id}"
+            self.kv.set(record_key, json.dumps(record))
+
+            # 更新表统计
+            self._update_table_stats(table_name, record_count_delta=1)
+
+            # 更新索引
+            self._update_indexes(table_name, record_id, data)
+
+            logger.debug(f"Inserted record {record_id} into table {table_name}")
+            return record_id
+
+        except Exception as e:
+            logger.error(f"Failed to insert record into table '{table_name}': {e}")
+            raise e
+
+    def get(self, table_name: str, record_id: str) -> Optional[Dict[str, Any]]:
+        """
+        获取单条记录
+
+        Args:
+            table_name: 表名
+            record_id: 记录ID
+
+        Returns:
+            记录数据，如果不存在返回None
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            record_key = f"{self.table_prefix}{table_name}:record:{record_id}"
+            record_str = self.kv.get(record_key)
+
+            if record_str:
+                record = json.loads(record_str)
+                return record
+
+            return None
+
+        except Exception as e:
+            logger.error(f"Failed to get record {record_id} from table '{table_name}': {e}")
+            raise e
+
+    def update(self, table_name: str, record_id: str, data: Dict[str, Any]) -> bool:
+        """
+        更新记录
+
+        Args:
+            table_name: 表名
+            record_id: 记录ID
+            data: 新的数据
+
+        Returns:
+            更新成功返回True
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            # 获取现有记录
+            existing_record = self.get(table_name, record_id)
+            if not existing_record:
+                logger.warning(f"Record {record_id} not found in table {table_name}")
+                return False
+
+            # 更新记录
+            updated_record = {
+                "id": record_id,
+                "data": data,
+                "created_at": existing_record.get("created_at"),
+                "updated_at": datetime.now().isoformat()
+            }
+
+            # 保存更新后的记录
+            record_key = f"{self.table_prefix}{table_name}:record:{record_id}"
+            self.kv.set(record_key, json.dumps(updated_record))
+
+            # 更新索引
+            self._update_indexes(table_name, record_id, data, existing_record.get("data"))
+
+            logger.debug(f"Updated record {record_id} in table {table_name}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to update record {record_id} in table '{table_name}': {e}")
+            raise e
+
+    def delete(self, table_name: str, record_id: str) -> bool:
+        """
+        删除记录
+
+        Args:
+            table_name: 表名
+            record_id: 记录ID
+
+        Returns:
+            删除成功返回True
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            # 获取记录以便清理索引
+            existing_record = self.get(table_name, record_id)
+            if not existing_record:
+                logger.warning(f"Record {record_id} not found in table {table_name}")
+                return False
+
+            # 删除记录
+            record_key = f"{self.table_prefix}{table_name}:record:{record_id}"
+            self.kv.remove(record_key)
+
+            # 更新表统计
+            self._update_table_stats(table_name, record_count_delta=-1)
+
+            # 清理索引
+            self._remove_from_indexes(table_name, record_id, existing_record.get("data"))
+
+            logger.debug(f"Deleted record {record_id} from table {table_name}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to delete record {record_id} from table '{table_name}': {e}")
+            raise e
+
+    def find_all(self, table_name: str, limit: int = 100) -> List[Dict[str, Any]]:
+        """
+        获取表中所有记录
+
+        Args:
+            table_name: 表名
+            limit: 最大返回记录数
+
+        Returns:
+            记录列表
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            # 获取所有记录键
+            record_keys = self._get_all_record_keys(table_name, limit)
+
+            if not record_keys:
+                return []
+
+            # 批量获取记录
+            records_data = self.kv.gets(record_keys)
+            records = []
+
+            for key, record_str in records_data.items():
+                if record_str:
+                    try:
+                        record = json.loads(record_str)
+                        records.append(record)
+                    except json.JSONDecodeError:
+                        logger.warning(f"Failed to parse record: {key}")
+
+            return records
+
+        except Exception as e:
+            logger.error(f"Failed to find all records in table '{table_name}': {e}")
+            raise e
+
+    def find_by_field(self, table_name: str, field: str, value: Any, limit: int = 100) -> List[Dict[str, Any]]:
+        """
+        根据字段值查找记录
+
+        Args:
+            table_name: 表名
+            field: 字段名
+            value: 字段值
+            limit: 最大返回记录数
+
+        Returns:
+            匹配的记录列表
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            # 简单的线性搜索（后续可以优化为索引搜索）
+            all_records = self.find_all(table_name, limit * 2)  # 获取更多记录以便过滤
+
+            matching_records = []
+            for record in all_records:
+                if len(matching_records) >= limit:
+                    break
+
+                record_data = record.get("data", {})
+                if record_data.get(field) == value:
+                    matching_records.append(record)
+
+            return matching_records
+
+        except Exception as e:
+            logger.error(f"Failed to find records by field in table '{table_name}': {e}")
+            raise e
+
+    def count(self, table_name: str) -> int:
+        """
+        获取表中记录数量
+
+        Args:
+            table_name: 表名
+
+        Returns:
+            记录数量
+        """
+        try:
+            if not self._table_exists(table_name):
+                raise ValueError(f"Table '{table_name}' does not exist")
+
+            table_meta = self._get_table_metadata(table_name)
+            return table_meta.get("record_count", 0)
+
+        except Exception as e:
+            logger.error(f"Failed to count records in table '{table_name}': {e}")
+            raise e
+
+    # 辅助方法
+    def _table_exists(self, table_name: str) -> bool:
+        """检查表是否存在"""
+        metadata = self._get_metadata()
+        return table_name in metadata.get("tables", [])
+
+    def _get_table_metadata(self, table_name: str) -> Dict[str, Any]:
+        """获取表元数据"""
+        table_meta_key = f"{self.meta_prefix}table:{table_name}"
+        meta_str = self.kv.get(table_meta_key)
+        if meta_str:
+            return json.loads(meta_str)
+        return {}
+
+    def _update_table_stats(self, table_name: str, record_count_delta: int = 0):
+        """更新表统计信息"""
+        try:
+            table_meta = self._get_table_metadata(table_name)
+            table_meta["record_count"] = table_meta.get("record_count", 0) + record_count_delta
+
+            table_meta_key = f"{self.meta_prefix}table:{table_name}"
+            self.kv.set(table_meta_key, json.dumps(table_meta))
+        except Exception as e:
+            logger.error(f"Failed to update table stats for '{table_name}': {e}")
+
+    def _get_all_record_keys(self, table_name: str, limit: int = 1000) -> List[str]:
+        """获取表中所有记录的键"""
+        try:
+            prefix = f"{self.table_prefix}{table_name}:record:"
+
+            # 使用KV的list_keys功能
+            result = self.kv.list_keys(prefix=prefix, limit=limit)
+
+            if result and "result" in result:
+                keys = []
+                for key_info in result["result"]:
+                    if isinstance(key_info, dict):
+                        keys.append(key_info.get("name"))
+                    else:
+                        keys.append(str(key_info))
+                return keys
+
+            return []
+
+        except Exception as e:
+            logger.error(f"Failed to get record keys for table '{table_name}': {e}")
+            return []
+
+    def _update_indexes(self, table_name: str, record_id: str, data: Dict[str, Any], old_data: Dict[str, Any] = None):
+        """更新索引（简单实现，后续可扩展）"""
+        # 这里可以实现索引更新逻辑
+        # 目前为简单实现，不做复杂索引
         pass
+
+    def _remove_from_indexes(self, table_name: str, record_id: str, data: Dict[str, Any]):
+        """从索引中移除记录"""
+        # 这里可以实现索引清理逻辑
+        pass
+
+    def _drop_table_indexes(self, table_name: str):
+        """删除表的所有索引"""
+        # 这里可以实现索引删除逻辑
+        pass
+
+    # 实用方法
+    def list_tables(self) -> List[str]:
+        """列出所有表"""
+        metadata = self._get_metadata()
+        return metadata.get("tables", [])
+
+    def table_info(self, table_name: str) -> Dict[str, Any]:
+        """获取表信息"""
+        if not self._table_exists(table_name):
+            raise ValueError(f"Table '{table_name}' does not exist")
+
+        return self._get_table_metadata(table_name)
+
+    def database_info(self) -> Dict[str, Any]:
+        """获取数据库信息"""
+        metadata = self._get_metadata()
+
+        # 添加表的详细信息
+        tables_info = {}
+        for table_name in metadata.get("tables", []):
+            tables_info[table_name] = self._get_table_metadata(table_name)
+
+        return {
+            "database_key": self.key,
+            "metadata": metadata,
+            "tables": tables_info
+        }
         

python_core/database/user.py

@@ -5,4 +5,5 @@ from .db import Db
 class UserDb(Db):
     def __init__(self):
         self.key = "model"
+        self.users = self.kv.get(self.key)
         pass