Session Manager¶

def to_dict(self) -> dict[str, str | None]:
    """Convert to dictionary for JSON serialization."""
    return {
        "doctor_id": self.doctor_id,
        "doctor_name": self.doctor_name,
        "eid": self.eid,
        "created_at": self.created_at,
        "expires_at": self.expires_at,
    }

@classmethod
def from_dict(cls, data: dict[str, str]) -> t.Self:
    """Create from dictionary."""
    return cls(
        doctor_id=data["doctor_id"],
        doctor_name=data["doctor_name"],
        eid=data["eid"],
        created_at=data["created_at"],
        expires_at=data["expires_at"],
    )

async def init(self) -> None:
    """Initialize the session manager.

    Creates session directory with restricted permissions and starts
    automatic cleanup task.
    """
    # Create session directory with restricted permissions
    self.session_dir.mkdir(mode=0o700, parents=True, exist_ok=True)

    # Start automatic cleanup task
    self._cleanup_task = asyncio.create_task(self._cleanup_loop())

    self.logger.info(
        "Session manager initialized",
        session_dir=str(self.session_dir),
        ttl_hours=self.ttl.total_seconds() / 3600,
    )

async def close(self) -> None:
    """Close the session manager.

    Stops the cleanup task. Session file is preserved for next
    startup.
    """
    if self._cleanup_task:
        self._cleanup_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._cleanup_task
        self._cleanup_task = None

    self.logger.info("Session manager closed")

async def login(
    self,
    doctor_id: str,
    eid: str,
    *,
    doctor_name: str | None = None,
    ttl: datetime.timedelta | None = None,
) -> SessionData:
    """Create a login session.

    If a session already exists, it will be replaced with the new one.
    Session persists across application restarts until expiration.

    Args:
        doctor_id: Doctor's unique identifier.
        doctor_name: Doctor's display name.
        eid: Doctor's eid.
        ttl: Time-to-live for the session. If None, uses default.

    Returns:
        Created SessionData.

    Example:
        ```python
        session = await manager.login(
            doctor_id="doctor-abc123",
            doctor_name="张医生",
            eid="dr_zhang",
        )
        print(f"Logged in as: {session.doctor_name}")
        ```
    """
    async with self._lock:
        ttl = ttl or self.ttl
        now = utils.utcnow()
        expires_at = now + ttl

        session_data = SessionData(
            doctor_id=doctor_id,
            doctor_name=doctor_name,
            eid=eid,
            created_at=now.isoformat(),
            expires_at=expires_at.isoformat(),
        )

        # Encrypt and write session data
        await self._write_session(session_data)

        self.logger.bind(
            doctor_id=doctor_id,
            eid=eid,
            expires_at=expires_at.isoformat(),
        ).info(f"Session created for {eid}")

        return session_data

async def logout(self) -> bool:
    """Logout by deleting the session.

    Returns:
        True if session was deleted, False if no session exists.

    Example:
        ```python
        logged_out = await manager.logout()
        if logged_out:
            print("Successfully logged out")
        ```
    """
    async with self._lock:
        if not self.session_file.exists():
            return False

        try:
            self.session_file.unlink()
            self.logger.info("Session deleted (logout)")
            return True
        except Exception as e:
            self.logger.error(f"Failed to delete session: {e}", exc_info=True)
            return False

async def is_logged_in(self) -> bool:
    """Check if there's an active (non-expired) session.

    Returns:
        True if there's an active session, False otherwise.

    Example:
        ```python
        if await manager.is_logged_in():
            print("User is logged in")
        else:
            print("Please login")
        ```
    """
    session = await self.get_session()
    return session is not None

async def get_session(self) -> SessionData | None:
    """Get current session if exists and not expired.

    Returns:
        SessionData if session exists and is valid, None otherwise.

    Example:
        ```python
        session = await manager.get_session()
        if session:
            print(f"Logged in as: {session.doctor_name}")
            print(f"Doctor ID: {session.doctor_id}")
        else:
            print("No active session")
        ```
    """
    async with self._lock:
        if not self.session_file.exists():
            return None

        try:
            # Read and decrypt session data
            session_data = await self._read_session()
            if session_data is None:
                return None

            # Check if expired
            expires_at = datetime.datetime.fromisoformat(session_data.expires_at)
            if utils.utcnow() >= expires_at:
                self.logger.debug("Session expired, removing file")
                self.session_file.unlink()
                return None

            return session_data

        except Exception as e:
            self.logger.warning(f"Failed to read session: {e}", exc_info=True)
            # If data is corrupted, remove file
            if self.session_file.exists():
                self.session_file.unlink()
            return None

async def get_doctor_id(self) -> str | None:
    """Get current logged-in doctor's ID.

    Returns:
        Doctor ID if logged in, None otherwise.

    Example:
        ```python
        doctor_id = await manager.get_doctor_id()
        if doctor_id:
            print(f"Current doctor: {doctor_id}")
        ```
    """
    session = await self.get_session()
    return session.doctor_id if session else None

async def extend_session(self, extra_ttl: datetime.timedelta) -> bool:
    """Extend current session expiration time.

    Args:
        extra_ttl: Additional time to add to expiration.

    Returns:
        True if session was extended, False if no session exists.

    Example:
        ```python
        # Extend by 2 hours
        extended = await manager.extend_session(timedelta(hours=2))
        if extended:
            print("Session extended")
        ```
    """
    async with self._lock:
        session = await self.get_session()
        if session is None:
            return False

        # Calculate new expiration
        current_expires = datetime.datetime.fromisoformat(session.expires_at)
        new_expires = current_expires + extra_ttl

        # Create updated session
        updated_session = SessionData(
            doctor_id=session.doctor_id,
            doctor_name=session.doctor_name,
            eid=session.eid,
            created_at=session.created_at,
            expires_at=new_expires.isoformat(),
        )

        await self._write_session(updated_session)
        self.logger.info(
            "Session extended",
            new_expires_at=new_expires.isoformat(),
        )

        return True