(self, path: Path)
| 551 | return logs[0].get("path") |
| 552 | |
| 553 | def _get_or_create(self, path: Path) -> LogSession: |
| 554 | key = str(path) |
| 555 | session = self.sessions.get(key) |
| 556 | if session: |
| 557 | return session |
| 558 | session = LogSession(path, self.socketio, self.shutdown_event) |
| 559 | session.load_initial() |
| 560 | self.sessions[key] = session |
| 561 | return session |
| 562 | |
| 563 | def subscribe(self, sid: str, path: Optional[str]) -> Optional[LogSession]: |
| 564 | with self.lock: |
no test coverage detected