astra.logger

Custom logging handler for observatory operations with database storage.

This module provides a specialized logging handler that extends Python’s standard logging.Handler to provide dual-output logging: console display and database storage. It’s designed specifically for observatory automation systems where logging events need to be both immediately visible and persistently stored for analysis.

Key features: - Dual logging output (console and database) - Error state tracking for the parent instance - Automatic timestamp formatting with microsecond precision - Exception and stack trace capture - SQL injection protection through quote escaping - UTC timezone standardization

The handler is particularly useful for long-running observatory operations where: - Real-time monitoring of system status is required - Historical log analysis is needed for debugging - Error states need to be tracked at the instance level - Database queries on log data are necessary

Typical usage: >>> from astra.logger import ObservatoryLogger, DatabaseLoggingHandler >>> from astra.database_manager import DatabaseManager >>> observatory_name = ‘MyObservatory’ >>> db_manager = DatabaseManager(observatory_name) >>> logger = ObservatoryLogger(observatory_name) >>> logger.addHandler(DatabaseLoggingHandler(db_manager))

Note

The handler expects the instance to have ‘error_free’ attribute and ‘cursor’ attribute.

Classes

ConsoleStreamHandler([log_traceback])	Initialize the handler.
CustomFormatter([fmt, datefmt])	A custom logging formatter that allows customizable formatting and color-coded output.
DatabaseLoggingHandler(database_manager)	Custom logging handler for dual-output to console and database.
DatabaseManagerProtocol(*args, **kwargs)
FileHandler(filename[, log_traceback])	Open the specified file and use it as the stream for logging.
ObservatoryLogger(name[, error_source, ...])	Custom logger for observatory operations with error tracking.

class astra.logger.DatabaseManagerProtocol(*args, **kwargs)

Bases: Protocol

execute(query: str) → Any

class astra.logger.ObservatoryLogger(name: str, error_source: list | None = None, error_free: bool = True, level=logging.INFO)

Bases: Logger

Custom logger for observatory operations with error tracking.

error_source

List to track sources of errors.

Type:

list

error_free

Flag indicating if the logger has encountered errors.

Type:

bool

error()

Overrides logging.Logger.error to set error_free to False.

critical()

Overrides logging.Logger.critical to set error_free to False.

report_device_issue()

Logs device-specific issues and tracks error sources.

Initialize the logger with a name and an optional level.

error(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘ERROR’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.error(“Houston, we have a %s”, “major problem”, exc_info=True)

critical(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘CRITICAL’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.critical(“Houston, we have a %s”, “major disaster”, exc_info=True)

report_device_issue(device_type: str, device_name: str, message: str, exception: Exception | None = None, exc_info: bool = True, level: Literal['error', 'warning'] = 'error') → None

class astra.logger.DatabaseLoggingHandler(database_manager: DatabaseManagerProtocol)

Bases: Handler

Custom logging handler for dual-output to console and database.

Extends Python’s standard logging.Handler to provide specialized logging for observatory automation systems. Simultaneously outputs log messages to console for real-time monitoring and stores them in database for persistent storage and analysis.

database_manager

Instance managing database operations, specifically database_manager.execute.

Type:

DatabaseManager

Initializes the instance - basically setting the formatter to None and the filter list to empty.

emit(record: LogRecord) → None

Process and emit a log record to console and database.

This method is called automatically by the logging framework when a log message is generated. It formats the record for console output, tracks error states in the parent instance, and stores the record in the database.

Parameters:

record – The log record to be processed and emitted.

Note

If the log level is ERROR or higher, sets instance.error_free to False. All log records are stored in the ‘log’ database table with timestamp, level, module, function, line number, and message.

class astra.logger.ConsoleStreamHandler(log_traceback: bool = True, **kwargs)

Bases: StreamHandler

Initialize the handler.

If stream is not specified, sys.stderr is used.

emit(record: LogRecord) → None

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

class astra.logger.FileHandler(filename: str | Path, log_traceback: bool = True, **kwargs)

Bases: FileHandler

Open the specified file and use it as the stream for logging.

FORMAT = '%(levelname)s,%(asctime)s.%(msecs)03d,%(process)d,%(name)s,(%(filename)s:%(lineno)d),%(message)s'

DATEFMT = '%Y-%m-%d %H:%M:%S'

emit(record: LogRecord) → None

Emit a record.

If the stream was not opened because ‘delay’ was specified in the constructor, open it before calling the superclass’s emit.

If stream is not open, current mode is ‘w’ and _closed=True, record will not be emitted (see Issue #42378).

class astra.logger.CustomFormatter(fmt=None, datefmt=None)

Bases: Formatter

A custom logging formatter that allows customizable formatting and color-coded output.

Uses ANSI escape codes to colorize log messages based on their severity level, if the console supports it.

Parameters

fmtstr, optional

The log message format. Defaults to ‘%(asctime)s :: %(levelname)-8s :: %(message)s’.

datefmtstr, optional

The date format for log timestamps. Defaults to ‘%H:%M:%S’.

Attributes

greystr

ANSI escape code for grey text.

greenstr

ANSI escape code for green text.

yellowstr

ANSI escape code for yellow text.

redstr

ANSI escape code for red text.

bold_redstr

ANSI escape code for bold red text.

resetstr

ANSI escape code to reset text formatting.

Methods

format(record)

Format the log record according to the specified log level’s formatting.

Usage

formatter = CustomFormatter(fmt=’%(asctime)s - %(levelname)s - %(message)s’, datefmt=’%Y-%m-%d %H:%M:%S’) handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler)

Note

The ANSI escape codes are used to colorize the output text in supported terminals.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

grey = '\x1b[38;20m'

green = '\x1b[32;20m'

yellow = '\x1b[33;20m'

red = '\x1b[31;20m'

bold_red = '\x1b[31;1m'

reset = '\x1b[0m'

format(record)

Format the log record according to the specified log level’s formatting.