Database API Reference
FlavumHive uses SQLite for persistent storage, managing platform states, tracking interactions, and storing analytics data. This document details the database API and schema.
Database Schema
Platform Actions
CREATE TABLE platform_actions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
action_type TEXT NOT NULL,
content TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
status TEXT NOT NULL,
metadata JSON
);
Rate Limits
CREATE TABLE rate_limits (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
action_type TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
window_start DATETIME NOT NULL,
window_end DATETIME NOT NULL,
count INTEGER DEFAULT 1
);
Personality States
CREATE TABLE personality_states (
id INTEGER PRIMARY KEY AUTOINCREMENT,
personality_name TEXT NOT NULL,
platform TEXT NOT NULL,
state JSON NOT NULL,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
Analytics
CREATE TABLE analytics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
metric_type TEXT NOT NULL,
value REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
metadata JSON
);
Database Manager
Class Definition
class DatabaseManager:
"""
Manages database operations for FlavumHive.
Attributes:
db_path (str): Path to SQLite database
connection (sqlite3.Connection): Database connection
logger (Logger): Logger instance
"""
def __init__(self, db_path: str):
self.db_path = db_path
self.connection = None
self.logger = logging.getLogger(__name__)
Core Methods
Initialize Database
async def initialize(self) -> bool:
"""
Initialize database and create tables.
Returns:
bool: Success status
"""
try:
self.connection = sqlite3.connect(self.db_path)
await self._create_tables()
return True
except Exception as e:
self.logger.error(f"Database initialization failed: {e}")
return False
Record Action
async def record_action(
self,
platform: str,
action_type: str,
content: Optional[str] = None,
metadata: Optional[Dict] = None
) -> int:
"""
Record a platform action.
Args:
platform: Platform name
action_type: Type of action
content: Optional action content
metadata: Optional metadata
Returns:
int: Action ID
"""
Check Rate Limits
async def check_rate_limit(
self,
platform: str,
action_type: str,
window_seconds: int
) -> bool:
"""
Check if action is within rate limits.
Args:
platform: Platform name
action_type: Type of action
window_seconds: Time window in seconds
Returns:
bool: True if action is allowed
"""
Platform State Management
Save Platform State
async def save_platform_state(
self,
platform: str,
state: Dict
) -> bool:
"""
Save platform state.
Args:
platform: Platform name
state: State data
Returns:
bool: Success status
"""
Load Platform State
async def load_platform_state(
self,
platform: str
) -> Optional[Dict]:
"""
Load platform state.
Args:
platform: Platform name
Returns:
Optional[Dict]: Platform state if exists
"""
Personality State Management
Save Personality State
async def save_personality_state(
self,
personality: str,
platform: str,
state: Dict
) -> bool:
"""
Save personality state.
Args:
personality: Personality name
platform: Platform name
state: State data
Returns:
bool: Success status
"""
Load Personality State
async def load_personality_state(
self,
personality: str,
platform: str
) -> Optional[Dict]:
"""
Load personality state.
Args:
personality: Personality name
platform: Platform name
Returns:
Optional[Dict]: Personality state if exists
"""
Analytics Management
Record Metric
async def record_metric(
self,
platform: str,
metric_type: str,
value: float,
metadata: Optional[Dict] = None
) -> bool:
"""
Record an analytics metric.
Args:
platform: Platform name
metric_type: Type of metric
value: Metric value
metadata: Optional metadata
Returns:
bool: Success status
"""
Get Analytics
async def get_analytics(
self,
platform: str,
metric_type: str,
start_time: datetime,
end_time: datetime
) -> List[Dict]:
"""
Get analytics data.
Args:
platform: Platform name
metric_type: Type of metric
start_time: Start of time range
end_time: End of time range
Returns:
List[Dict]: Analytics data
"""
Maintenance Operations
Cleanup Old Data
async def cleanup_old_data(
self,
days_to_keep: int = 30
) -> bool:
"""
Clean up old data.
Args:
days_to_keep: Days of data to retain
Returns:
bool: Success status
"""
Optimize Database
async def optimize(self) -> bool:
"""
Optimize database.
Returns:
bool: Success status
"""
Error Handling
Database Errors
class DatabaseError(Exception):
"""Base class for database errors."""
pass
class ConnectionError(DatabaseError):
"""Database connection errors."""
pass
class QueryError(DatabaseError):
"""Query execution errors."""
pass
Usage Examples
Basic Usage
# Initialize database
db = DatabaseManager("bot.db")
await db.initialize()
# Record action
action_id = await db.record_action(
platform="twitter",
action_type="tweet",
content="Hello, World!",
metadata={"reply_to": "123456"}
)
# Check rate limits
can_tweet = await db.check_rate_limit(
platform="twitter",
action_type="tweet",
window_seconds=3600
)
State Management
# Save state
state = {
"last_tweet_id": "123456",
"session_data": {"cookies": {...}}
}
await db.save_platform_state("twitter", state)
# Load state
state = await db.load_platform_state("twitter")
Analytics
# Record metrics
await db.record_metric(
platform="twitter",
metric_type="engagement_rate",
value=0.85,
metadata={"tweet_id": "123456"}
)
# Get analytics
analytics = await db.get_analytics(
platform="twitter",
metric_type="engagement_rate",
start_time=datetime.now() - timedelta(days=7),
end_time=datetime.now()
)
Best Practices
-
Connection Management
- Use context managers
- Close connections properly
- Handle timeouts
-
Error Handling
- Catch specific exceptions
- Log errors appropriately
- Implement retries
-
Performance
- Use transactions
- Index key columns
- Regular maintenance
-
Data Integrity
- Validate inputs
- Use constraints
- Regular backups