Add core modules (session, recorder, database)
This commit is contained in:
88
.termflow/core/session.py
Normal file
88
.termflow/core/session.py
Normal file
@@ -0,0 +1,88 @@
|
||||
"""Session data model."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
from dateutil import parser
|
||||
|
||||
|
||||
@dataclass
|
||||
class Command:
|
||||
"""Represents a single command in a session."""
|
||||
command: str
|
||||
timestamp: str = ""
|
||||
exit_code: Optional[int] = None
|
||||
output: Optional[str] = None
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"command": self.command,
|
||||
"timestamp": self.timestamp,
|
||||
"exit_code": self.exit_code,
|
||||
"output": self.output,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Command":
|
||||
return cls(
|
||||
command=data.get("command", ""),
|
||||
timestamp=data.get("timestamp", ""),
|
||||
exit_code=data.get("exit_code"),
|
||||
output=data.get("output"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Session:
|
||||
"""Represents a recorded terminal session."""
|
||||
name: str
|
||||
id: Optional[int] = None
|
||||
start_time: Optional[datetime] = None
|
||||
end_time: Optional[datetime] = None
|
||||
commands: List[Command] = field(default_factory=list)
|
||||
created_at: Optional[datetime] = None
|
||||
command_count: int = 0
|
||||
|
||||
def __post_init__(self):
|
||||
if self.command_count == 0 and self.commands:
|
||||
self.command_count = len(self.commands)
|
||||
|
||||
def add_command(self, command: Command) -> None:
|
||||
"""Add a command to the session."""
|
||||
self.commands.append(command)
|
||||
self.command_count += 1
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"start_time": self.start_time.isoformat() if self.start_time else None,
|
||||
"end_time": self.end_time.isoformat() if self.end_time else None,
|
||||
"commands": [cmd.to_dict() for cmd in self.commands],
|
||||
"created_at": self.created_at.isoformat() if self.created_at else None,
|
||||
"command_count": self.command_count,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Session":
|
||||
commands = [Command.from_dict(cmd) for cmd in data.get("commands", [])]
|
||||
start_time = data.get("start_time")
|
||||
end_time = data.get("end_time")
|
||||
created_at = data.get("created_at")
|
||||
|
||||
if isinstance(start_time, str):
|
||||
start_time = parser.parse(start_time)
|
||||
if isinstance(end_time, str):
|
||||
end_time = parser.parse(end_time)
|
||||
if isinstance(created_at, str):
|
||||
created_at = parser.parse(created_at)
|
||||
|
||||
return cls(
|
||||
id=data.get("id"),
|
||||
name=data.get("name", ""),
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
commands=commands,
|
||||
created_at=created_at,
|
||||
command_count=data.get("command_count", len(commands)),
|
||||
)
|
||||
Reference in New Issue
Block a user