Core Utilities

lariv.utils

Core utilities and helper functions for string formatting, database connection parsing, migration operations, and subprocess execution.

AddFieldToOtherApp

Bases: Operation

Add a field to a model in another app (used by connector apps).

Source code in lariv/utils.py

class AddFieldToOtherApp(migrations.operations.base.Operation):
    """Add a field to a model in another app (used by connector apps)."""

    reversible = True
    reduces_to_sql = True

    def __init__(
        self,
        target_app_label,
        model_name,
        name,
        field,
        preserve_default=True,
    ):
        self.target_app_label = target_app_label
        self.model_name = model_name
        self.model_name_lower = model_name.lower()
        self.name = name
        self.field = field
        self.preserve_default = preserve_default

    def state_forwards(self, app_label, state):
        state.add_field(
            self.target_app_label,
            self.model_name_lower,
            self.name,
            self.field,
            self.preserve_default,
        )

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        to_model = to_state.apps.get_model(self.target_app_label, self.model_name)
        if self.allow_migrate_model(schema_editor.connection.alias, to_model):
            from_model = from_state.apps.get_model(
                self.target_app_label, self.model_name
            )
            field = to_model._meta.get_field(self.name)
            if not self.preserve_default:
                field.default = self.field.default
            schema_editor.add_field(from_model, field)
            if not self.preserve_default:
                field.default = NOT_PROVIDED

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        from_model = from_state.apps.get_model(self.target_app_label, self.model_name)
        if self.allow_migrate_model(schema_editor.connection.alias, from_model):
            schema_editor.remove_field(
                from_model, from_model._meta.get_field(self.name)
            )

AlterFieldInOtherApp

Bases: Operation

Alter a field on a model in another app (used by connector apps).

Source code in lariv/utils.py

class AlterFieldInOtherApp(migrations.operations.base.Operation):
    """Alter a field on a model in another app (used by connector apps)."""

    reversible = True
    reduces_to_sql = True

    def __init__(
        self, target_app_label, model_name, name, field, preserve_default=True
    ):
        self.target_app_label = target_app_label
        self.model_name = model_name
        self.model_name_lower = model_name.lower()
        self.name = name
        self.field = field
        self.preserve_default = preserve_default

    def state_forwards(self, app_label, state):
        state.alter_field(
            self.target_app_label,
            self.model_name_lower,
            self.name,
            self.field,
            self.preserve_default,
        )

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        to_model = to_state.apps.get_model(self.target_app_label, self.model_name)
        if self.allow_migrate_model(schema_editor.connection.alias, to_model):
            from_model = from_state.apps.get_model(
                self.target_app_label, self.model_name
            )
            from_field = from_model._meta.get_field(self.name)
            to_field = to_model._meta.get_field(self.name)
            if not self.preserve_default:
                to_field.default = self.field.default
            schema_editor.alter_field(from_model, from_field, to_field)
            if not self.preserve_default:
                to_field.default = NOT_PROVIDED

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        self.database_forwards(app_label, schema_editor, from_state, to_state)

convert_m2m(v)

Convert a Django ManyRelatedManager or similar iterable to a standard Python list.

Parameters:

Name	Type	Description	Default
v		The value to convert.	required

Returns:

Name	Type	Description
list		A list representation of the provided value.

Source code in lariv/utils.py

def convert_m2m(v):
    """
    Convert a Django ManyRelatedManager or similar iterable to a standard Python list.

    Args:
        v: The value to convert.

    Returns:
        list: A list representation of the provided value.
    """
    if hasattr(v, "all"):
        return list(v.all())
    return v

format_datetime(dt, fmt='%B %d, %Y at %I:%M %p')

Format a datetime object into a readable string using the server's local timezone.

Parameters:

Name	Type	Description	Default
dt	datetime	The datetime to format.	required
fmt	str	The strftime format to use.	'%B %d, %Y at %I:%M %p'

Returns:

Name	Type	Description
str	str	The formatted datetime string.

Source code in lariv/utils.py

def format_datetime(dt, fmt="%B %d, %Y at %I:%M %p") -> str:
    """
    Format a datetime object into a readable string using the server's local timezone.

    Args:
        dt (datetime): The datetime to format.
        fmt (str): The strftime format to use.

    Returns:
        str: The formatted datetime string.
    """
    if not dt:
        return ""
    local_dt = timezone.localtime(dt)
    return local_dt.strftime(fmt)

get_db_uri(connection_alias='default')

Constructs an SQLAlchemy-compatible database URI from Django's DATABASES setting.

Parameters:

Name	Type	Description	Default
connection_alias	str	The Django database alias to resolve. Defaults to "default".	'default'

Returns:

Name	Type	Description
str	str	The constructed connection URI (e.g., postgresql://...).

Source code in lariv/utils.py

def get_db_uri(connection_alias="default") -> str:
    """
    Constructs an SQLAlchemy-compatible database URI from Django's DATABASES setting.

    Args:
        connection_alias (str): The Django database alias to resolve. Defaults to "default".

    Returns:
        str: The constructed connection URI (e.g., `postgresql://...`).
    """
    db_config = settings.DATABASES[connection_alias]
    engine = db_config["ENGINE"].split(".")[-1]

    # Mapping Django backends to standard URI schemes
    scheme_map = {
        "postgresql": "postgres",
        "postgis": "postgres",
        "mysql": "mysql",
        "sqlite3": "sqlite",
        "oracle": "oracle",
    }
    scheme = scheme_map.get(engine, engine)

    # Handle SQLite separately (file path, no host/user)
    if scheme == "sqlite":
        db_path = db_config["NAME"]
        return f"sqlite:///{db_path}"

    # Handle standard Client/Server databases
    user = quote(db_config.get("USER", ""))
    password = quote(db_config.get("PASSWORD", ""))
    host = quote(db_config.get("HOST", "localhost"))
    port = db_config.get("PORT", "")
    name = quote(db_config.get("NAME", ""))

    # Construct the authority part (user:pass@host:port)
    netloc = f"{user}:{password}@{host}"
    if port:
        netloc += f":{port}"

    return f"{scheme}://{netloc}/{name}"

pascalcase(input)

Convert a snake_case or space-separated string into PascalCase.

Parameters:

Name	Type	Description	Default
input	str	The string to format.	required

Returns:

Name	Type	Description
str	str	The PascalCase string.

Source code in lariv/utils.py

def pascalcase(input: str) -> str:
    """
    Convert a snake_case or space-separated string into PascalCase.

    Args:
        input (str): The string to format.

    Returns:
        str: The PascalCase string.
    """
    output = ""
    if len(input) > 0:
        output += input[0].upper()
        input = input[1:]

    newWord = False
    lastWord = ""
    for char in input:
        if (not lastWord.isupper() and char.isupper()) or newWord:
            newWord = False
            output += f"{char.upper()}"
            continue
        if char in "_- ":
            newWord = True
            continue
        output += char
    return output

run_exe(command, args, cwd, env, wait_for_end=False)

Runs an executable as a subprocess cleanly handling paths and environments.

Parameters:

Name	Type	Description	Default
command	str	the name of the executable to run (resolved via shutil.which).	required
args	list[str]	list of arguments to be passed into the command.	required
cwd	str	absolute path for the working directory of the process.	required
env	dict[str, str]	additional environment variables beyond os.environ.	required
wait_for_end	bool	whether to block and wait for execution to complete or spawn detached.	False

Raises:

Type	Description
ValueError	In case the command binary could not be found via the system PATH.
Exception	Re-raises any exceptions generated by subprocess.Popen.

Source code in lariv/utils.py

def run_exe(command: str, args: list[str], cwd: str, env: dict[str, str], wait_for_end: bool = False):
    """
    Runs an executable as a subprocess cleanly handling paths and environments.

    Args:
        command (str): the name of the executable to run (resolved via `shutil.which`).
        args (list[str]): list of arguments to be passed into the command.
        cwd (str): absolute path for the working directory of the process.
        env (dict[str, str]): additional environment variables beyond `os.environ`.
        wait_for_end (bool): whether to block and wait for execution to complete or spawn detached.

    Raises:
        ValueError: In case the command binary could not be found via the system PATH.
        Exception: Re-raises any exceptions generated by `subprocess.Popen`.
    """
    bin = shutil.which(command)
    process_env = os.environ.copy()
    process_env.update(env)
    if not bin:
        raise ValueError(f"Couldn't find the {command} binary")

    process = subprocess.Popen(
        [bin, *args],
        cwd=cwd,
        env=process_env,
        start_new_session=True,
    )

    if wait_for_end:
        process.wait()

    if not wait_for_end:
        def kill_service():
            try:
                process.kill()
                process.wait()
            except Exception as e:
                logging.error(f"{e}")

        atexit.register(kill_service)

table_page_range(page)

Calculate windowed pagination ranges for rendering table footers (e.g., [1, 2, 3, 4] or [[1], [4,5,6], [10]]).

Parameters:

Name	Type	Description	Default
page		A Django pagination Page object.	required

Returns:

Type	Description
list[list[int]]	list[list[int]]: A list of page number blocks used to render pagination gaps.

Source code in lariv/utils.py

def table_page_range(page) -> list[list[int]]:
    """
    Calculate windowed pagination ranges for rendering table footers (e.g., [1, 2, 3, 4] or [[1], [4,5,6], [10]]).

    Args:
        page: A Django pagination `Page` object.

    Returns:
        list[list[int]]: A list of page number blocks used to render pagination gaps.
    """
    num_pages = page.paginator.num_pages
    if num_pages == 0:
        return []

    current = page.number
    low = max(1, current - WINDOW)
    high = min(num_pages, current + WINDOW)

    main_start = 1 if low <= 2 else low
    main_end = num_pages if high >= num_pages - 1 else high

    out = []

    if main_start > 1:
        out.append([1])

    out.append(list(range(main_start, main_end + 1)))

    if main_end < num_pages:
        out.append([num_pages])

    return out

titlecase(input)

Convert a snake_case, camelCase, or pascalCase string into space-separated Title Case.

Parameters:

Name	Type	Description	Default
input	str	The string to format.	required

Returns:

Name	Type	Description
str	str	The Title Cased string.

Source code in lariv/utils.py

def titlecase(input: str) -> str:
    """
    Convert a snake_case, camelCase, or pascalCase string into space-separated Title Case.

    Args:
        input (str): The string to format.

    Returns:
        str: The Title Cased string.
    """
    output = ""
    if len(input) > 0:
        output += input[0].upper()
        input = input[1:]

    newWord = False
    lastWord = ""
    for char in input:
        if (not lastWord.isupper() and char.isupper()) or newWord:
            newWord = False
            output += f" {char.upper()}"
            continue
        if char in "_- ":
            newWord = True
            continue
        output += char
    return output

update_url(url, **kwargs)

Update or append URL query parameters safely.

Parameters:

Name	Type	Description	Default
url	str	The original URL.	required
**kwargs		The query string parameters to add or override.	{}

Returns:

Name	Type	Description
str	str	The updated URL.

Source code in lariv/utils.py

def update_url(url: str, **kwargs) -> str:
    """
    Update or append URL query parameters safely.

    Args:
        url (str): The original URL.
        **kwargs: The query string parameters to add or override.

    Returns:
        str: The updated URL.
    """
    url_parts = url.split("?")
    path = url_parts[0]
    query_str = url_parts[1] if len(url_parts) > 1 else ""
    query_params = parse_qs(query_str)
    query_params.update(kwargs)
    return path + "?" + urlencode(query_params, doseq=True)