dan daaniam

Using MongoDB BSON ObjectId with Pydantic (FastAPI, pymongo or motor)

The important part here is when_used="json". This allows the serialization of data (ObjectId) as a string with FastAPI, while also ensuring that model_dump() retains the ObjectId for MongoDB. Not just for _id, but for every reference with ObjectId.

PyObjectId

from typing import Annotated, Any
from bson import ObjectId

Alembic

Custom file name template

# alembic.ini
file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s

Overide sqlalchemy database url from config file

Ruff file watchers in PyCharm + basic commands

Format code file watcher:

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

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 \

Randoms

const _ = require('lodash');

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

Mount pgadmin4 settings with passwords (example)

Dirs structure

project_root
 - docker
   - pgadmin4
     - pgpass
     - servers.json


	FIREBASE_JWKS_PUBLIC_KEYS = 'https://www.googleapis.com/robot/v1/metadata/x509/securetoken@system.gserviceaccount.com'

	keys_cache = {'keys': None, 'last_fetched': 0.0}


	async def fetch_public_keys() -> dict[str, str]:
	"""Fetch Firebase public keys from URL, with caching for 1 hour."""

	# Check if the cache is older than 1 hour (3600 seconds)


	# Basic configuration
	logging.basicConfig(
	level=logging.DEBUG,
	format="%(asctime)s - %(levelname)s - %(name)s - %(funcName)20s() on line %(lineno)s - %(message)s",
	# datefmt="%Y-%m-%d %H:%M:%S.%f %Z%z",
	handlers=[stream_handler],
	)

	"""
	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()

	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)