diff --git a/python_core/database/db.py b/python_core/database/db.py index a27f74c..5e9ddee 100644 --- a/python_core/database/db.py +++ b/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 + } \ No newline at end of file diff --git a/python_core/database/user.py b/python_core/database/user.py index 78e583a..601501b 100644 --- a/python_core/database/user.py +++ b/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 \ No newline at end of file