Core

Shared base classes and core abstractions: logging, string utilities, SQL safety.

Core package — lazy-loaded.

Contains logging, string utilities, SQL safety, and other core functions. All submodules load on first attribute access via PEP 562 __getattr__.

siege_utilities.core.get_logger(name=None)[source]

Get a logger instance.

Parameters:

name (str | None)

Return type:

Logger

siege_utilities.core.log_info(message, logger_name=None)[source]

Log an info message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.log_warning(message, logger_name=None)[source]

Log a warning message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.log_error(message, logger_name=None)[source]

Log an error message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.log_debug(message, logger_name=None)[source]

Log a debug message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.log_critical(message, logger_name=None)[source]

Log a critical message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.init_logger(name, log_to_file=False, log_dir='logs', level='INFO', max_bytes=5000000, backup_count=5, shared_log_file=None)[source]

Initialize a logger with specific configuration.

Deprecated since version The: name suggests this function does first-time setup distinct from get_logger(), but both are get-or-create with shared config. Prefer configure_shared_logging() to set the shared config and get_logger() to retrieve loggers; the init_logger shape will be removed in a future release.

Parameters:
  • name (str) – Logger name

  • log_to_file (bool) – Whether to log to file

  • log_dir (str | Path) – Directory for log files

  • level (str | int) – Log level

  • max_bytes (int) – Max file size before rotation

  • backup_count (int) – Number of backup files

  • shared_log_file (str | Path | None) – Path to shared log file

Returns:

Configured logger instance

Return type:

Logger

siege_utilities.core.configure_shared_logging(log_file_path=None, level='INFO', max_bytes=5000000, backup_count=5)[source]

Configure shared logging for all loggers.

Parameters:
  • log_file_path (str | Path | None) – Path to shared log file. If None, configures console-only logging.

  • level (str | int) – Log level for logging

  • max_bytes (int) – Max file size before rotation (only used with file logging)

  • backup_count (int) – Number of backup files to keep (only used with file logging)

Return type:

None

siege_utilities.core.remove_wrapping_quotes_and_trim(target_string)[source]

Removes wrapping quotes (single or double) from a string and trims whitespace

Parameters:

target_string (str) – String that may have wrapping quotes and whitespace

Returns:

String with any wrapping quotes and whitespace removed

Return type:

str

siege_utilities.core.clean_string(target_string)[source]

Clean a string by removing quotes, trimming whitespace, and normalizing.

Parameters:

target_string (str | None) – String to clean

Returns:

Cleaned string

Return type:

str

siege_utilities.core.normalize_whitespace(target_string)[source]

Normalize whitespace in a string by replacing multiple spaces with single space.

Parameters:

target_string (str | None) – String to normalize

Returns:

String with normalized whitespace

Return type:

str

Examples

>>> normalize_whitespace("hello    world")
'hello world'
>>> normalize_whitespace("test\n\n\nvalue")
'test value'
siege_utilities.core.snake_case(text)[source]

Convert text to snake_case.

Parameters:

text (str) – Text to convert

Returns:

snake_case version of text

Return type:

str

Examples

>>> snake_case("HelloWorld")
'hello_world'
>>> snake_case("hello-world")
'hello_world'
>>> snake_case("Hello World")
'hello_world'
siege_utilities.core.remove_non_alphanumeric(text, keep_chars='_-')[source]

Remove non-alphanumeric characters from text, optionally keeping specified chars.

Parameters:
  • text (str) – Text to clean

  • keep_chars (str) – Characters to keep (default: underscore and hyphen)

Returns:

Cleaned text

Return type:

str

Examples

>>> remove_non_alphanumeric("hello@world!")
'helloworld'
>>> remove_non_alphanumeric("test_value-123", keep_chars="_-")
'test_value-123'
siege_utilities.core.validate_sql_identifier(name, label='identifier', *, allow_dotted=False)[source]

Validate that name is a safe SQL identifier.

Checks that the identifier contains only alphanumeric characters and underscores, starts with a letter or underscore, and is non-empty. This prevents SQL injection when identifiers must be interpolated into SQL strings (e.g., CREATE DATABASE {name}).

Parameters:
  • name (str) – The identifier to validate (database name, table name, etc.).

  • label (str) – Human-readable label for error messages (e.g., “database”, “table”).

  • allow_dotted (bool) – If True, accept schema.table or catalog.schema.table – every dot-separated component must itself be a safe identifier. Defaults to False to preserve backwards-compatible single-component behaviour.

Returns:

The validated identifier (unchanged).

Raises:

ValueError – If name is empty, structurally invalid, or (when allow_dotted=False) contains a dot.

Return type:

str

Examples

>>> validate_sql_identifier("electronic_silver", "database")
'electronic_silver'
>>> validate_sql_identifier("silver; DROP TABLE --", "database")
Traceback (most recent call last):
    ...
ValueError: Invalid SQL database: ...
>>> validate_sql_identifier("public.users", "table", allow_dotted=True)
'public.users'
siege_utilities.core.validate_sql_identifier_in(name, allowed, label='identifier')[source]

Validate that name is both structurally safe AND in allowed.

Use this when the caller has a known set of legal values – the typical case is a column name that must exist on a DataFrame:

validate_sql_identifier_in(geometry_col, df.columns, "column")
Raises:

ValueErrorname is structurally unsafe OR not in allowed.

Parameters:
Return type:

str

siege_utilities.core.escape_sql_string_literal(value, *, dialect='standard')[source]

SQL-escape a string for use inside single-quoted SQL literals.

Use sparingly – parameter binding is always preferred. The cases that legitimately need this are:

  • Sedona / Spark-SQL spatial UDFs (ST_GeomFromText) where Spark parameter substitution doesn’t apply uniformly across versions.

  • Hand-written SQL files generated for human inspection / batch re-import – defensive escaping in case the row data is later hand-edited.

The escape rule is the SQL standard: replace ' with ''. For MySQL (dialect="mysql"), backslashes are also doubled because MySQL treats \ as an escape character by default (NO_BACKSLASH_ESCAPES off).

NUL bytes are rejected – most drivers refuse to send them anyway and they can split a query in C-string-aware backends.

Parameters:
  • value (str) – The string to escape.

  • dialect (str) – "standard" (default) or "mysql".

Raises:
  • TypeErrorvalue is not a string.

  • ValueErrorvalue contains a NUL byte or dialect is unknown.

Return type:

str

siege_utilities.core.validate_sql_fragment(fragment, label='SQL fragment')[source]

Reject SQL fragments containing dangerous structural keywords.

This is defense-in-depth for cases where a free-form SQL expression (e.g. a WHERE clause) must be interpolated into a query string. Parameter binding is always preferred; use this only when binding is not possible.

The blocklist catches statement-level injection (semicolons, DDL/DML keywords, comments). It does NOT make arbitrary input safe — callers should still avoid passing untrusted user input.

Parameters:
  • fragment (str) – The SQL fragment to validate.

  • label (str) – Human-readable label for error messages.

Returns:

The validated fragment (unchanged).

Raises:
Return type:

str

Examples

>>> validate_sql_fragment("state_fips = '06'")
"state_fips = '06'"
>>> validate_sql_fragment("1=1; DROP TABLE users--")
Traceback (most recent call last):
    ...
ValueError: Dangerous pattern in SQL fragment: ...

Submodules

Modern logging system for siege_utilities. Provides structured logging with proper configuration management.

siege_utilities.core.logging.LoggerName

alias of str

class siege_utilities.core.logging.LoggerManager[source]

Bases: object

Manages multiple loggers with consistent configuration.

__init__()[source]

Initialize the logger manager.

configure_shared_logging(log_file_path=None, level='INFO', max_bytes=5000000, backup_count=5)[source]

Configure shared logging for all loggers.

Parameters:
  • log_file_path (str | Path | None) – Path to shared log file. If None, configures console-only logging.

  • level (str | int) – Log level for shared file (or console if no file)

  • max_bytes (int) – Max file size before rotation

  • backup_count (int) – Number of backup files to keep

Return type:

None

get_logger(name=None)[source]

Get or create a logger with the specified name.

Parameters:

name (str | None) – Logger name, uses default if None

Returns:

Configured logger instance

Return type:

Logger

cleanup_logger(name)[source]

Remove and cleanup a logger.

Parameters:

name (str)

Return type:

bool

cleanup_all_loggers()[source]

Cleanup all loggers.

Return type:

None

set_default_logger_name(name)[source]

Set the default logger name.

Parameters:

name (str)

Return type:

None

class siege_utilities.core.logging.LoggingConfig[source]

Bases: object

Configuration for logging system.

log_to_file: bool = False
log_dir: Path
max_bytes: int = 5000000
backup_count: int = 5
log_to_console: bool = True
console_level: str | int = 'INFO'
file_level: str | int = 'DEBUG'
shared_log_file: Path | None = None
shared_level: str | int = 'INFO'
__init__(log_to_file=False, log_dir=<factory>, max_bytes=5000000, backup_count=5, log_to_console=True, console_level='INFO', file_level='DEBUG', shared_log_file=None, shared_level='INFO')
Parameters:
Return type:

None

siege_utilities.core.logging.configure_shared_logging(log_file_path=None, level='INFO', max_bytes=5000000, backup_count=5)[source]

Configure shared logging for all loggers.

Parameters:
  • log_file_path (str | Path | None) – Path to shared log file. If None, configures console-only logging.

  • level (str | int) – Log level for logging

  • max_bytes (int) – Max file size before rotation (only used with file logging)

  • backup_count (int) – Number of backup files to keep (only used with file logging)

Return type:

None

siege_utilities.core.logging.get_logger(name=None)[source]

Get a logger instance.

Parameters:

name (str | None)

Return type:

Logger

siege_utilities.core.logging.init_logger(name, log_to_file=False, log_dir='logs', level='INFO', max_bytes=5000000, backup_count=5, shared_log_file=None)[source]

Initialize a logger with specific configuration.

Deprecated since version The: name suggests this function does first-time setup distinct from get_logger(), but both are get-or-create with shared config. Prefer configure_shared_logging() to set the shared config and get_logger() to retrieve loggers; the init_logger shape will be removed in a future release.

Parameters:
  • name (str) – Logger name

  • log_to_file (bool) – Whether to log to file

  • log_dir (str | Path) – Directory for log files

  • level (str | int) – Log level

  • max_bytes (int) – Max file size before rotation

  • backup_count (int) – Number of backup files

  • shared_log_file (str | Path | None) – Path to shared log file

Returns:

Configured logger instance

Return type:

Logger

siege_utilities.core.logging.cleanup_logger(name)[source]

Cleanup a specific logger.

Parameters:

name (str)

Return type:

bool

siege_utilities.core.logging.cleanup_all_loggers()[source]

Cleanup all loggers.

Return type:

None

siege_utilities.core.logging.set_default_logger_name(name)[source]

Set the default logger name.

Parameters:

name (str)

Return type:

None

siege_utilities.core.logging.log_debug(message, logger_name=None)[source]

Log a debug message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.logging.log_info(message, logger_name=None)[source]

Log an info message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.logging.log_warning(message, logger_name=None)[source]

Log a warning message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.logging.log_error(message, logger_name=None)[source]

Log an error message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.logging.log_critical(message, logger_name=None)[source]

Log a critical message.

Parameters:
  • message (str)

  • logger_name (str | None)

Return type:

None

siege_utilities.core.logging.parse_log_level(level)[source]

Convert a string or numeric level into a logging level constant.

Parameters:

level (str | int) – String (‘DEBUG’, ‘INFO’, etc.) or int (10, 20, etc.)

Returns:

Logging level constant

Return type:

int

siege_utilities.core.logging.temporary_logging_config(config)[source]

Temporarily change logging configuration.

Parameters:

config (LoggingConfig)

Return type:

Generator[None, None, None]

siege_utilities.core.string_utils.remove_wrapping_quotes_and_trim(target_string)[source]

Removes wrapping quotes (single or double) from a string and trims whitespace

Parameters:

target_string (str) – String that may have wrapping quotes and whitespace

Returns:

String with any wrapping quotes and whitespace removed

Return type:

str

siege_utilities.core.string_utils.clean_string(target_string)[source]

Clean a string by removing quotes, trimming whitespace, and normalizing.

Parameters:

target_string (str | None) – String to clean

Returns:

Cleaned string

Return type:

str

siege_utilities.core.string_utils.normalize_whitespace(target_string)[source]

Normalize whitespace in a string by replacing multiple spaces with single space.

Parameters:

target_string (str | None) – String to normalize

Returns:

String with normalized whitespace

Return type:

str

Examples

>>> normalize_whitespace("hello    world")
'hello world'
>>> normalize_whitespace("test\n\n\nvalue")
'test value'
siege_utilities.core.string_utils.snake_case(text)[source]

Convert text to snake_case.

Parameters:

text (str) – Text to convert

Returns:

snake_case version of text

Return type:

str

Examples

>>> snake_case("HelloWorld")
'hello_world'
>>> snake_case("hello-world")
'hello_world'
>>> snake_case("Hello World")
'hello_world'
siege_utilities.core.string_utils.remove_non_alphanumeric(text, keep_chars='_-')[source]

Remove non-alphanumeric characters from text, optionally keeping specified chars.

Parameters:
  • text (str) – Text to clean

  • keep_chars (str) – Characters to keep (default: underscore and hyphen)

Returns:

Cleaned text

Return type:

str

Examples

>>> remove_non_alphanumeric("hello@world!")
'helloworld'
>>> remove_non_alphanumeric("test_value-123", keep_chars="_-")
'test_value-123'

SQL identifier validation + literal escaping for Hive/Spark/PostGIS/DuckDB/Snowflake.

Prevents SQL injection where identifiers (database/table/column names) must be interpolated into SQL strings – most dialects don’t permit parameter-bound identifiers, so we validate against a strict allow-list instead. For value escaping, prefer parameter binding; this module’s escape_sql_string_literal() is the fallback for cases where the underlying API genuinely cannot bind (Sedona spatial UDFs, hand-written SQL files for re-import, etc.).

Originally developed for pure-translation (electinfo/enterprise). Moved to siege_utilities as shared security-critical infrastructure.

siege_utilities.core.sql_safety.validate_sql_identifier(name, label='identifier', *, allow_dotted=False)[source]

Validate that name is a safe SQL identifier.

Checks that the identifier contains only alphanumeric characters and underscores, starts with a letter or underscore, and is non-empty. This prevents SQL injection when identifiers must be interpolated into SQL strings (e.g., CREATE DATABASE {name}).

Parameters:
  • name (str) – The identifier to validate (database name, table name, etc.).

  • label (str) – Human-readable label for error messages (e.g., “database”, “table”).

  • allow_dotted (bool) – If True, accept schema.table or catalog.schema.table – every dot-separated component must itself be a safe identifier. Defaults to False to preserve backwards-compatible single-component behaviour.

Returns:

The validated identifier (unchanged).

Raises:

ValueError – If name is empty, structurally invalid, or (when allow_dotted=False) contains a dot.

Return type:

str

Examples

>>> validate_sql_identifier("electronic_silver", "database")
'electronic_silver'
>>> validate_sql_identifier("silver; DROP TABLE --", "database")
Traceback (most recent call last):
    ...
ValueError: Invalid SQL database: ...
>>> validate_sql_identifier("public.users", "table", allow_dotted=True)
'public.users'
siege_utilities.core.sql_safety.validate_sql_identifier_in(name, allowed, label='identifier')[source]

Validate that name is both structurally safe AND in allowed.

Use this when the caller has a known set of legal values – the typical case is a column name that must exist on a DataFrame:

validate_sql_identifier_in(geometry_col, df.columns, "column")
Raises:

ValueErrorname is structurally unsafe OR not in allowed.

Parameters:
Return type:

str

siege_utilities.core.sql_safety.escape_sql_string_literal(value, *, dialect='standard')[source]

SQL-escape a string for use inside single-quoted SQL literals.

Use sparingly – parameter binding is always preferred. The cases that legitimately need this are:

  • Sedona / Spark-SQL spatial UDFs (ST_GeomFromText) where Spark parameter substitution doesn’t apply uniformly across versions.

  • Hand-written SQL files generated for human inspection / batch re-import – defensive escaping in case the row data is later hand-edited.

The escape rule is the SQL standard: replace ' with ''. For MySQL (dialect="mysql"), backslashes are also doubled because MySQL treats \ as an escape character by default (NO_BACKSLASH_ESCAPES off).

NUL bytes are rejected – most drivers refuse to send them anyway and they can split a query in C-string-aware backends.

Parameters:
  • value (str) – The string to escape.

  • dialect (str) – "standard" (default) or "mysql".

Raises:
  • TypeErrorvalue is not a string.

  • ValueErrorvalue contains a NUL byte or dialect is unknown.

Return type:

str

siege_utilities.core.sql_safety.validate_sql_fragment(fragment, label='SQL fragment')[source]

Reject SQL fragments containing dangerous structural keywords.

This is defense-in-depth for cases where a free-form SQL expression (e.g. a WHERE clause) must be interpolated into a query string. Parameter binding is always preferred; use this only when binding is not possible.

The blocklist catches statement-level injection (semicolons, DDL/DML keywords, comments). It does NOT make arbitrary input safe — callers should still avoid passing untrusted user input.

Parameters:
  • fragment (str) – The SQL fragment to validate.

  • label (str) – Human-readable label for error messages.

Returns:

The validated fragment (unchanged).

Raises:
Return type:

str

Examples

>>> validate_sql_fragment("state_fips = '06'")
"state_fips = '06'"
>>> validate_sql_fragment("1=1; DROP TABLE users--")
Traceback (most recent call last):
    ...
ValueError: Dangerous pattern in SQL fragment: ...