monitor_mail ¶

Save into '/usr/local/bin/monitor_mail', then:

chmod +x /usr/local/bin/monitor_mail

#!/usr/bin/env python3
import time
import re
import argparse

# Strong styles for primary fields
STATUS_STYLES = {
    'sent':      '\033[97;42m',  # white on green
    'bounced':   '\033[97;41m',  # white on red
    'deferred':  '\033[30;44m',  # black on blue
    'expired':   '\033[97;45m',  # white on magenta
    'reject':    '\033[97;41m',  # white on red
    'discarded': '\033[30;47m',  # black on white
    'hold':      '\033[30;46m',  # black on cyan
    'error':     '\033[97;41m',  # white on red
}
DEFAULT_STYLE = '\033[30;43m'    # black on yellow
SECONDARY_STYLE = '\033[94m'     # light blue
RESET = '\033[0m'

def highlight_status_and_to(line, use_color=True):
    """
    Highlights:
    - 'status=...' and 'to=<...>' with background color
    - 'from=...', 'relay=...', 'message-id=...' with light foreground color
    Skips orig_to.
    """
    status_match = re.search(r'status=([a-zA-Z0-9_]+)', line)
    if not status_match:
        return None

    status_value = status_match.group(1)

    if use_color:
        strong_style = STATUS_STYLES.get(status_value, DEFAULT_STYLE)
        secondary_style = SECONDARY_STYLE
    else:
        strong_style = ''
        secondary_style = ''

    reset = RESET if use_color else ''

    # Primary highlights
    line = re.sub(
        r'status=([a-zA-Z0-9_]+)',
        lambda m: f"{strong_style}{m.group(0)}{reset}",
        line
    )
    line = re.sub(
        r'(?<!orig_)to=<[^>]+>',
        lambda m: f"{strong_style}{m.group(0)}{reset}",
        line
    )

    # Secondary highlights
    line = re.sub(
        r'from=<[^>]+>',
        lambda m: f"{secondary_style}{m.group(0)}{reset}",
        line
    )
    line = re.sub(
        r'relay=[^ ,]+',
        lambda m: f"{secondary_style}{m.group(0)}{reset}",
        line
    )
    line = re.sub(
        r'message-id=<[^>]+>',
        lambda m: f"{secondary_style}{m.group(0)}{reset}",
        line
    )

    return line

def tail_f(filepath, show_all=False, from_start=False, use_color=True, only_status=None):
    """
    Tail -f with optional filtering, coloring and status filtering.
    """
    with open(filepath, 'r') as f:
        if not from_start:
            f.seek(0, 2)
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.1)
                continue
            line = line.rstrip()

            # Extract status value (if any)
            status_match = re.search(r'status=([a-zA-Z0-9_]+)', line)
            status_value = status_match.group(1) if status_match else None

            if only_status and status_value != only_status:
                continue  # Skip line if it doesn't match the --only-status filter

            highlighted = highlight_status_and_to(line, use_color=use_color)
            if highlighted:
                print(highlighted)
            elif show_all:
                print(line)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Tail and highlight Postfix log lines")
    parser.add_argument(
        "--show-all", action="store_true",
        help="Show all lines (not just those with 'status=...')"
    )
    parser.add_argument(
        "--from-start", action="store_true",
        help="Start reading from the beginning of the file instead of the end"
    )
    parser.add_argument(
        "--no-color", action="store_true",
        help="Disable all ANSI color output (monochrome)"
    )
    parser.add_argument(
        "--only-status", metavar="STATUS",
        help="Show only lines with a specific status (e.g. bounced, sent, deferred)"
    )

    args = parser.parse_args()

    # Default path to Postfix log
    log_file_path = "/var/log/mail.log"

    # Start tailing
    tail_f(
        filepath=log_file_path,
        show_all=args.show_all,
        from_start=args.from_start,
        use_color=not args.no_color,
        only_status=args.only_status
    )

  List Prometheus labels and metrics ¶

file "list_promethues_metrics_and_labels.py"

#!/usr/bin/env python3
import requests
import json
import argparse

PROM_URL = "http://localhost:9090"


def list_metrics_with_labels():
    data = {}
    r = requests.get(f"{PROM_URL}/api/v1/label/__name__/values")
    metric_names = r.json()["data"]
    for name in sorted(metric_names):
        r2 = requests.get(f"{PROM_URL}/api/v1/series", params={"match[]": name})
        series = r2.json()["data"]
        label_set = set()
        for s in series:
            label_set.update(s.keys())
        label_set.discard("__name__")
        data[name] = list(label_set)
    return data


if __name__ == "__main__":

    parser = argparse.ArgumentParser()
    parser.add_argument('-f', '--filter', type=str, help="filter keys")
    parser.add_argument('-m', '--metrics-only', action='store_true', help="List only metrics (exclude labels)")
    parser.add_argument('-i', '--indent', type=int, default=4, help="No indent if < 0")
    args = parser.parse_args()

    data = list_metrics_with_labels()

    if args.filter:
        data = {
            k: v
            for k, v in data.items()
            if args.filter in k
        }

    indent = args.indent
    if indent < 0:
        indent = None

    if args.metrics_only:
        print(json.dumps(list(data.keys()), indent=indent))
    else:
        print(json.dumps(data, indent=indent))

  SEO Tips for Multi-Tenant Django Applications ¶

  Export Gitlab Issue ¶

#!/usr/bin/env python3
import pprint
import json
import gitlab
import argparse

# Requirements
# python-gitlab==4.11.1

GITLAB_URL = "https://gitlab.brainstorm.it"
GITLAB_PRIVATE_TOKEN = "glpat-******************"
GITLAB_PROJECT = 77
PAGE_BREAK_STRING = '<div style="page-break-after: always;"></div>'


class GitlabClient(gitlab.Gitlab):

    def __init__(self, project_id):
        super().__init__(GITLAB_URL, GITLAB_PRIVATE_TOKEN, order_by='created_at')
        self.project = self.projects.get(project_id)

    def retrieve_issue(self, issue_id):

        issue = self.project.issues.get(issue_id)

        base_url = issue.web_url
        # Es: 'https://gitlab.brainstorm.it/group/project/-/issues/36'
        position = base_url.find('/-/')
        if position >= 0:
            base_url = base_url[:position]
        # Es: 'https://gitlab.brainstorm.it/group/project'

        data = json.loads(issue.to_json())
        self.fix_image_links(data, base_url)
        data['notes'] = []

        notes = issue.notes.list(all=True)
        notes = sorted([n for n in notes if not n.system], key=lambda k: k.created_at)

        for n in notes:
            note = json.loads(n.to_json())
            self.fix_image_links(note, base_url)
            data['notes'].append(note)

        return data

    def fix_image_links(self, data, base_url):
        text = data.get('description', data.get('body'))
        text = text.replace('/uploads/', base_url+'/uploads/')
        if 'description' in data:
            data['description'] = text
        else:
            data['body'] = text

def to_markdown(data):
    text = ""
    for k, v in data.items():

        text += '\n# [%d] %s\n\n' % (k, v['title'])
        text += "### %s (%s)\n\n" % (v['author']['name'], v['created_at'])
        text += "**Link**: %s\n\n" % v['web_url']
        text += v['description']
        text += "\n\n"

        for n in v['notes']:

            text += "\n\n" + (80*'-') + "\n\n"
            text += "### %s (%s)\n\n" % (n['author']['name'], v['created_at'])
            text += n['body']
            text += "\n\n"

        text += "\n\n" + PAGE_BREAK_STRING

    return text


def main():
    parser = argparse.ArgumentParser(description="...")
    parser.add_argument('issue_ids', nargs="+", type=int)
    parser.add_argument("--project_id", "-p", type=int, default=GITLAB_PROJECT, help="Default: %d" % GITLAB_PROJECT)
    parser.add_argument("--format", type=str, choices=["", "markdown", ], default="")
    args = parser.parse_args()

    project_id = args.project_id
    client = GitlabClient(GITLAB_PROJECT)
    data = {}
    for issue_id in args.issue_ids:
        issue_data = client.retrieve_issue(issue_id)
        data[issue_id] = issue_data

    if args.format == 'markdown':
        print(to_markdown(data))
    else:
        pprint.pprint(data)


if __name__ == '__main__':
    main()

  Debug python with rpdb ¶

.

pip install rpdb

import rpdb; rpdb.set_trace()

or

rpdb.set_trace(addr=('0.0.0.0', 4444))

nc <target_ip> 4444

  django-ninja Notes ¶

from ninja import Router
v1_router = Router(tags=["cclib"])


@v1_router.get("/hello")
def hello(request):
    return "Hello world!"

from typing import List
import uuid
from django.shortcuts import get_object_or_404
from ninja import Router
from ninja import Schema, ModelSchema
from ninja.security import django_auth

from cclib.resources.models import Document

v1_router = Router(tags=["cclib"])


class DocumentSchema(ModelSchema):
    class Meta:
        model = Document
        fields = "__all__"


@v1_router.get("/documents/{uuid:document_id}", response=DocumentSchema, auth=django_auth)
def get_document(request, document_id: uuid.UUID):
    return get_object_or_404(Document, pk=document_id)


@v1_router.get("/documents", response=List[DocumentSchema], auth=django_auth)
def list_documents(request):
    qs = Document.objects.all()
    return qs

import traceback
from typing import List
from ninja import Router
from ninja import Schema, ModelSchema
from ninja.errors import HttpError
from ninja.security import django_auth
from ninja.security import django_auth_superuser
from django.conf import settings
from django.contrib.auth.models import Group
from django.contrib.auth import get_user_model

v1_router = Router(tags=["ccserver"])


class GroupSchema(ModelSchema):
    class Meta:
        model = Group
        fields = "__all__"

class UserSchema(ModelSchema):
    # See: "Nested objects"
    # https://django-ninja.dev/guides/response/#nested-objects
    groups: List[GroupSchema] = []
    class Meta:
        model = get_user_model()
        #fields = "__all__"
        fields = [
            'id',
            'username',
        ]


@v1_router.get("/users", response=List[UserSchema], auth=django_auth_superuser)
def list_users(request):
    try:
        User = get_user_model()
        qs = User.objects.all()
    except Exception as e:
        # See: "Throwing HTTP responses with exceptions"
        # https://django-ninja.dev/guides/errors/#throwing-http-responses-with-exceptions
        raise HttpError(500, str(e) + "\n" + traceback.format_exc())

    return qs

from ninja import NinjaAPI
from ninja.security import django_auth

api = NinjaAPI(csrf=True)

@api.get("/pets", auth=django_auth)
def pets(request):
    return f"Authenticated user {request.auth}"

from ninja.security import HttpBearer


class AuthBearer(HttpBearer):
    def authenticate(self, request, token):
        if token == "supersecret":
            return token


@api.get("/bearer", auth=AuthBearer())
def bearer(request):
    return {"token": request.auth}

from ninja.errors import HttpError

@api.get("/some/resource")
def some_operation(request):
    if True:
        raise HttpError(503, "Service Unavailable. Please retry later.")

  PostgreSQL materialized views with Django ¶

My implementation:

import csv
from django.db import models
from django.db import connection
from django.utils import timezone
from django.db import connection
from django.contrib.humanize.templatetags.humanize import intcomma
from ccms_lib.backend.models import CODE_MAX_LENGTH


class SqlViewMixin():

    @classmethod
    def create_view(cls, verbose=False):
        view_name = cls._meta.db_table
        cls_sql = cls.sql().strip()
        if cls_sql.endswith(';'):
            cls_sql = cls_sql[:-1]

        if cls.materialized:
            sql = f"CREATE materialized VIEW IF NOT EXISTS {view_name} AS "
            sql += cls_sql + ';\n'
            pk_name = cls._meta.pk.name
            sql += f'CREATE UNIQUE INDEX IF NOT EXISTS idx_{view_name} ON {view_name}({pk_name});'
        else:
            sql = f"CREATE VIEW {view_name} AS "
            sql += cls_sql + ';\n'
        cls.execute(sql, verbose)

    @classmethod
    def drop_view(cls, verbose=False):
        view_name = cls._meta.db_table
        sql = "DROP %sVIEW IF EXISTS %s CASCADE;\n" % (
            'materialized ' if cls.materialized else '',
            view_name,
        )
        pk_name = cls._meta.pk.name
        sql += f"DROP INDEX IF EXISTS idx_{view_name};"
        cls.execute(sql, verbose)

    @classmethod
    def refresh_view(cls, concurrently=False, verbose=False):
        if cls.materialized:
            #sql = "REFRESH MATERIALIZED VIEW CONCURRENTLY {cls._meta.db_table};"
            sql = "REFRESH MATERIALIZED VIEW %s%s;" % (
                "CONCURRENTLY " if concurrently else '',
                cls._meta.db_table,
            )
            cls.execute(sql, verbose)

    @classmethod
    def execute(cls, sql, verbose):
        with connection.cursor() as cursor:
            if verbose:
                print(sql)
            cursor.execute(sql)

    @classmethod
    def export_view_as_csv(cls, verbose, filename=None, delimiter=','):

        def export_rows(cursor, sql, offset, page_size, writer, verbose):
            if page_size > 0:
                sql = sql + " OFFSET %d LIMIT %d" % (offset, page_size)
            if verbose:
                print(sql)
            cursor.execute(sql)
            if offset <= 0:
                writer.writerow([f.name for f in cursor.description])
            for row in cursor.fetchall():
                writer.writerow(row)

        view_name = cls._meta.db_table

        if filename is None:
            filename = timezone.now().strftime('%Y-%m-%d_%H-%M-%S__') + view_name + '.csv'

        #page_size = 100000 # 0 = no pagination
        page_size = 1000000 # 0 = no pagination
        sql = str(cls.objects.all().query)
        sql += ' ORDER BY "%s"' % cls._meta.pk.name
        n = 0
        step = 0

        if verbose:
            n = cls.objects.count()
            step = int(n / 100)
            print('Exporting file "%s"; records: %s' % (filename, intcomma(n)))

        with open(filename, 'w', newline='') as csvfile:
            writer = csv.writer(csvfile,  dialect='excel', delimiter=delimiter)
            with connection.cursor() as cursor:

                if page_size <= 0:
                    # no pagination
                    cursor.execute(sql)
                    writer.writerow([f.name for f in cursor.description])
                    j = 0
                    row = cursor.fetchone()
                    while row is not None:
                        j += 1
                        if verbose and j==1 or ((j/10) % step) == 0:
                            progress = int((j * 100) / n) + 1
                            print('%d%% (%s/%s)' % (progress, intcomma(j), intcomma(n)))
                        writer.writerow(row)
                        row = cursor.fetchone()

                else:
                    # paginate
                    num_pages = (n // page_size) + (1 if n % page_size else 0)
                    offset = 0
                    j = 0
                    while offset < n:

                        if verbose:
                            progress = int(((j*page_size) * 100) / n)
                            print('page %d/%d (%d%%)' % (j+1, num_pages, progress))

                        export_rows(cursor, sql, offset, page_size, writer, verbose=False)
                        offset += page_size
                        j += 1

################################################################################
# Example ...

class DwhBase(SqlViewMixin, models.Model):

    materialized = True

    id = models.CharField(null=False, max_length=256, primary_key=True)
    base_code = models.CharField(max_length=CODE_MAX_LENGTH, null=False)
    base_name = models.CharField(max_length=256, null=False)

    class Meta:
        managed = False
        db_table = "dwhm_base"

    def __str__(self):
        return str(self.id)

    @staticmethod
    def sql():
        return """
SELECT code AS id,
    code as base_code,
    description AS base_name
   FROM backend_base;
"""

References:

• Using PostgreSQL Views in Django
• Materialized Views for Faster Reads

  Download a large binary file in chunks with Python ¶

Using requests:

import requests

# https://stackoverflow.com/questions/16694907/download-large-file-in-python-with-requests#16696317
def download_file(url):
    local_filename = url.split('/')[-1]
    # NOTE the stream=True parameter below
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(local_filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192):
                # If you have chunk encoded response uncomment if
                # and set chunk_size parameter to None.
                #if chunk:
                f.write(chunk)
    return local_filename

Using urlib.request:

import urllib.request

def download_file2(url):
    """
    Use urlib.request instead of requests to avoid "403 Forbidden" errors
    when the remote server tries to prevent scraping (for example when
    is protected behind cloudflare)
    """
    request = urllib.request.Request(url)
    request.add_header('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:106.0) Gecko/20100101 Firefox/106.0')
    request.add_header('Accept', 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8')
    request.add_header('Accept-Language', 'en-US,en;q=0.5')

    response = urllib.request.urlopen(request)
    local_filename = url.split('/')[-1]
    chunk_size = 8192
    with open(local_filename, 'wb') as f:
        size = 0
        while True:
            info = response.read(chunk_size)
            if len(info) < 1:
                break
            size = size + len(info)
            f.write(info)
    return local_filename

  faker list ¶

from faker import Faker
faker = Faker()
names = dir(faker)

for name in names:
    try:
        value = getattr(faker, name)()
        text = str(value)[:40]
        print('%-30.30s %s' % (name, text))
    except:
        pass