daaniam

class AsyncTyper(typer.Typer):
    def __init__(self, *args, **kwargs):
        kwargs["no_args_is_help"] = True

        super().__init__(*args, **kwargs)

    def async_command(self, *args, **kwargs):
        def decorator(async_func):
            @wraps(async_func)
            def sync_func(*_args, **_kwargs):

daaniam

import re

from sqlalchemy.orm import DeclarativeBase, declared_attr


def resolve_table_name(name):
    """Resolve table name from Class CamelCase"""

    # Match a letter just before an upper-case char
    # Example: "Hello World" would match "o" because next "W" is uppercase.

daaniam

class MyCUstomResponse(JSONResponse):
    def __init__(
        self,
        data: list[t.Any],
        errors: list[t.Any] | None = None,
        status_code: int = 200,
        headers: dict[str, str] | None = None,
        media_type: str | None = None,
        background: BackgroundTask | None = None,
    ) -> None:

daaniam

"""
Short example of how to set realtionship in many-to-many with extra column(s) in association table.
creator class is used on the relationship(), but it's also possible to create MUserPermitAssoc
explicitly and pass it to MUser.permits.

"""

from __future__ import annotations

import sqlalchemy as sa

daaniam

Mount pgadmin4 settings with passwords (example)

Dirs structure

project_root
 - docker
   - pgadmin4
     - pgpass
     - servers.json

daaniam

Randoms

const _ = require('lodash');

// Function to generate random string
function randStr(length) {
    return _.join(_.times(length, () => _.random(35).toString(36)), '');
}

daaniam

Dockerfile

FROM debian:bookworm-slim
RUN apt update && apt install -y lftp

Send from terminal

docker run -v $(pwd)/file_to_send:/tmp/file_to_send \

daaniam

Ruff file watchers in PyCharm + basic commands

Format code file watcher:

Name: ruff-format
Program: $PyInterpreterDirectory$/ruff
Arguments: format .
Working directory: $ProjectFileDir$

daaniam

import asyncio

from sqlalchemy import ForeignKey, UniqueConstraint, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectin_polymorphic

aengine = create_async_engine("sqlite+aiosqlite:///database2.db")
asession = async_sessionmaker(aengine)

daaniam

  """
  Just a note where to find original asyncpg exception object wrapped by SQLAlchemy
  """
  
  
  try:
      created_record = await db.scalar(insert(Model).values(**data_in.model_dump()).returning(Model))
  except IntegrityError as err:
      if isinstance(err.orig.__cause__, UniqueViolationError):
          raise RecordAlreadyExists()