​
GalileoLoggerSingleton Objects

class GalileoLoggerSingleton()
A singleton class that manages a collection of GalileoLogger instances. This class ensures that only one instance exists across the application and provides a thread-safe way to retrieve or create GalileoLogger clients based on the given ‘project’ and ‘log_stream’ parameters. If the parameters are not provided, the class attempts to read the values from the environment variables GALILEO_PROJECT and GALILEO_LOG_STREAM. The loggers are stored in a dictionary using a tuple (project, log_stream) as the key.

​
get

def get(*,
        project: Optional[str] = None,
        log_stream: Optional[str] = None,
        experiment_id: Optional[str] = None,
        mode: str = "batch",
        local_metrics: Optional[list[LocalMetricConfig]] = None
        ) -> GalileoLogger
Retrieve an existing GalileoLogger or create a new one if it does not exist. This method first computes the key from the project and log_stream parameters, checks if a logger exists in the cache, and if not, creates a new GalileoLogger. The creation and caching are done in a thread-safe manner. Arguments:
  • project (Optional[str]): The project name. Defaults to None.
  • log_stream (Optional[str]): The log stream name. Defaults to None.
  • experiment_id (Optional[str]): The experiment ID. Defaults to None.
  • local_metrics (Optional[list[LocalScorerConfig]]): Local scorers to run on traces/spans. Only used if initializing a new logger, ignored otherwise. Defaults to None.
Returns: GalileoLogger: An instance of GalileoLogger corresponding to the key.

​
reset

def reset(project: Optional[str] = None,
          log_stream: Optional[str] = None,
          experiment_id: Optional[str] = None,
          mode: str = "batch") -> None
Reset (terminate and remove) one or all GalileoLogger instances. Arguments:
  • project (Optional[str]): The project name. Defaults to None.
  • log_stream (Optional[str]): The log stream name. Defaults to None.
  • experiment_id (Optional[str]): The experiment ID. Defaults to None.

​
reset_all

def reset_all() -> None
Reset (terminate and remove) all GalileoLogger instances.

​
flush

def flush(project: Optional[str] = None,
          log_stream: Optional[str] = None,
          experiment_id: Optional[str] = None,
          mode: str = "batch") -> None
Flush (upload and clear) a GalileoLogger instance. If both project and log_stream are None, then all cached loggers are flushed and cleared. Otherwise, only the specific logger corresponding to the provided key (project, log_stream) is flushed and removed. Arguments:
  • project (Optional[str]): The project name. Defaults to None.
  • log_stream (Optional[str]): The log stream name. Defaults to None.
  • experiment_id (Optional[str]): The experiment ID. Defaults to None.

​
flush_all

def flush_all() -> None
Flush (upload and clear) all GalileoLogger instances.

​
get_all_loggers

def get_all_loggers() -> dict[tuple[str, str, str, str], GalileoLogger]
Retrieve a copy of the dictionary containing all active loggers. Returns: Dict[Tuple[str, str, str], GalileoLogger]: A dictionary mapping keys (project, log_stream) to their corresponding GalileoLogger instances.