Reference library

Python Code Samples

Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.

4 matches
Automation & scripting easy

Create Mock Watermarked Image Bytes in Python Without PIL

Builds a mock image-like byte stream with an embedded watermark using only stdlib modules, for testing pipelines without PIL.

watermark bytes zlib
Python
from io import BytesIO
import zlib
import struct


def create_watermarked_bytes(width: int, height: int, watermark: bytes) -> bytes:
    """Create a mock image-like byte stream with a watermark (no PIL)."""
    header = struct.pack("<2I", width, height)
    payload = watermark * max(1, (width * height // max(1, len(wa…
13 0 Open
Data pipelines & processing easy

How to Implement Incremental Load with Watermark by updated_at in Python

Load only new or changed rows into SQLite by comparing an updated_at timestamp against a stored watermark, returning counts and the new watermark.

incremental-load watermark sqlite
Python
import sqlite3
from datetime import datetime, timedelta


def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
    """Load only rows with updated_at greater than the last watermark."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Create table if it doesn't exist
  …
12 0 Open
Streaming & messaging medium

Mock Watermark Late Event Side Output in Python

Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.

watermark streaming side output
Python
from datetime import datetime, timedelta
from typing import List, Tuple


def watermark_mock(
    events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
    """Simulate watermarking: events arriving on time vs. late by ch…
11 0 Open
Big data & Spark medium

How to Implement a Streaming Watermark in Python

Mock structured streaming watermarks in Python to track late event times and compute a watermark for windowed processing.

streaming watermark spark
Python
from datetime import datetime, timedelta
import time

class StreamingWatermark:
    """Mock watermark tracker for structured streaming."""

    def __init__(self, watermark_delay_seconds):
        self.watermark_delay = timedelta(seconds=watermark_delay_seconds)
        self.max_event_time = None

    def observe_even…
14 0 Open

Browse by section

Each section groups closely related Python snippets.

Guide: free Python code samples library

Copy-ready Python snippets for learners and developers

PythonSkillset code samples are short, focused examples organised by topic and difficulty. Every snippet is server-rendered HTML — readable by search engines and easy to copy. Open any sample, read the notes, copy the code, then press Try in editor to run it in the browser with Pyodide.

How to use this library

  1. Pick a topic section — strings, lists, files, functions, and more
  2. Open a sample, read How it works, and copy the code block
  3. Run it in the IDE, tweak values, then take a related quiz or tutorial lesson

Samples vs tutorials and challenges

Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.