Main SNI modules

Entry point

Main module

This module contains the entry point to SNI.

sni.__main__.configure_logging() → None

Basic configuration of the logging facility

sni.__main__.connect_database_signals() → None

Imports all signals submodules.

sni.__main__.main()

Entry point.

sni.__main__.migrate_database() → None

Inits and migrate the MongoDB database.

sni.__main__.parse_command_line_arguments() → argparse.Namespace

Parses command line arguments

See also

argparse documentation

sni.__main__.schedule_jobs() → None

Schedules jobs from all subpackages

sni.__main__.start_discord_bot() → None

Starts the discord bot (or rather, schedules it to be started), and corrects the Discord command and event handlers.

Configuration module

Configuration facility

class sni.conf.Config

Bases: pydantic.env_settings.BaseSettings

SNI configuration model

database: DatabaseConfig = None

discord: DiscordConfig = None

esi: ESIConfig = None

general: GeneralConfig = None

jwt: JWTConfig = None

redis: RedisConfig = None

sentry: SentryConfig = None

teamspeak: TeamspeakConfig = None

class sni.conf.DatabaseConfig

Bases: pydantic.main.BaseModel

Database configuration model

authentication_source: str = None

database: str = None

host: str = None

password: pdt.SecretStr = None

port: int = None

username: str = None

class sni.conf.DiscordConfig

Bases: pydantic.main.BaseModel

Discord configuration model

auth_channel_id: int = None

enabled: bool = None

log_channel_id: int = None

server_id: int = None

token: pdt.SecretStr = None

class sni.conf.ESIConfig

Bases: pydantic.main.BaseModel

ESI configuration model

client_id: pdt.SecretStr = None

client_secret: pdt.SecretStr = None

class sni.conf.GeneralConfig

Bases: pydantic.main.BaseModel

General configuration model

class LoggingLevel

Bases: str, enum.Enum

Acceptable logging levels

CRITICAL = 'critical'

DEBUG = 'debug'

ERROR = 'error'

INFO = 'info'

WARNING = 'warning'

debug: bool = None

host: Union[IPv4Address, IPv6Address] = None

logging_level: LoggingLevel = None

property logging_level_int

Returns the int equivalent of the logging_level field.

port: int = None

root_url: pdt.HttpUrl = None

scheduler_thread_count: int = None

class sni.conf.JWTConfig

Bases: pydantic.main.BaseModel

JWT configuration model

class JWTAlgorithm

Bases: str, enum.Enum

Acceptable JWT algorithms, see here.

HS256 = 'HS256'

HMAC using SHA-256 hash algorithm (default)

HS384 = 'HS384'

HMAC using SHA-384 hash algorithm

HS512 = 'HS512'

HMAC using SHA-512 hash algorithm

algorithm: JWTAlgorithm = None

secret: pdt.SecretBytes = None

class sni.conf.RedisConfig

Bases: pydantic.main.BaseModel

Redis configuration model

database: int = None

host: str = None

port: int = None

class sni.conf.SentryConfig

Bases: pydantic.main.BaseModel

Sentry configuration model.

See also

Sentry homepage <https://sentry.io/welcome/> Sentry documentation for Python ASGI

dsn: pdt.HttpUrl = None

enabled: bool = None

traces_sample_rate: float = None

class sni.conf.TeamspeakConfig

Bases: pydantic.main.BaseModel

Teamspeak configuration model

auth_group_name: str = None

bot_name: str = None

enabled: bool = None

host: str = None

password: pdt.SecretStr = None

port: int = None

server_id: int = None

username: str = None

sni.conf.load_configuration_file(path: str) → None

Loads the configuration from a YAML file.

sni.conf.print_configuration_schema(indentation: Optional[int] = 4) → None

Prints the JSON schema of sni.conf.Config

Scheduler

Asyncronous / concurrent / schedules job management.

Simply use the global member sni.scheduler.scheduler.

See also

APScheduler documentation

sni.scheduler.JOBS_KEY: str = 'scheduler:default:jobs'

The redis key for the job list

sni.scheduler.RUN_TIMES_KEY: str = 'scheduler:default:run_times'

The redis key for the job run times

sni.scheduler._test_tick() → None

Test function to check that the scheduler is really running.

Schedule like this:

scheduler.add_job(_test_tick, 'interval', seconds=3, jitter=0)

sni.scheduler._test_tock() → None

Test function to check that the scheduler is really running.

sni.scheduler.run_scheduled(function: Callable) → Callable

Decorator that makes a function scheduled to run immediately when called.

Example:

@signals.post_save.connect_via(User)
@run_scheduled
def test(_sender: Any, **kwargs):
    usr = kwargs['document']
    status = 'created' if kwargs.get('created', False) else 'updated'
    logging.debug('User %s has been %s', status, usr.character_name)

sni.scheduler.start_scheduler() → None

Clears the job store and starts the scheduler.

sni.scheduler.stop_scheduler() → None

Stops the scheduler and cleans up things

Other utilities

Various utilities

sni.utils.catch_all(function: Callable, error_message: str, *, args: Tuple[Any, ...] = (), kwargs: dict = {}) → None

Calls a function but catches all the exceptions. If any were raised, logs an error message, followed by the string representation of the exception.

async sni.utils.catch_all_async(function: Callable, error_message: str, *, args: Tuple[Any, ...] = (), kwargs: dict = {}) → None

Calls a function but catches all the exceptions. If any were raised, logs an error message, followed by the string representation of the exception.

sni.utils.catches_all(error_message: str = 'Error') → Callable

Decorator version of sni.utils.catch_all().

sni.utils.from_timestamp(timestamp: int) → datetime.datetime

Returns a UTC datetime from a UNIX timestamp.

sni.utils.now() → datetime.datetime

Returns the current UTC datetime.

sni.utils.now_plus(**kwargs) → datetime.datetime

Returns the current UTC datetime plus a specified timedelta.

See also

datetime.timedelta

sni.utils.object_from_name(name: str) → Any

Returns a callable from its name, e.g. sni.esi.jobs:refresh_tokens.

sni.utils.random_code(length: int) → str

Returns a random string made of digits, lowercase letters, and uppercase letters, of a given length.

Warning

Not cryptographically secure.