Signals

import secrets
import time

from flask import (
    Flask, appcontext_popped, appcontext_pushed, appcontext_tearing_down,
    flash, g, jsonify, render_template_string, request, request_finished,
    request_started, request_tearing_down,
)

app = Flask(__name__)
app.config.update(SECRET_KEY=secrets.token_hex(), TESTING=False)

BOOKS = {1: {"id": 1, "title": "Dune"}, 2: {"id": 2, "title": "Emma"}}
events: list[str] = []


@app.get("/books")
def book_list():
    flash("Listing books", "info")
    return jsonify(list(BOOKS.values()))


@app.get("/books/<int:book_id>/page")
def book_page(book_id):
    book = BOOKS.get(book_id) or ("missing", 404)
    return render_template_string("<h1>{{ book['title'] }}</h1>", book=book)


@request_started.connect_via(app)
def on_started(sender, **extra):
    g.started = time.perf_counter()
    events.append(f"request_started {request.method} {request.path}")


@request_finished.connect_via(app)
def on_finished(sender, response, **extra):
    elapsed_ms = (time.perf_counter() - g.started) * 1000
    events.append(f"request_finished {response.status_code}")
    assert elapsed_ms >= 0


@request_tearing_down.connect_via(app)
def on_request_down(sender, exc=None, **extra):
    events.append(f"request_tearing_down exc={exc!r}")


@appcontext_tearing_down.connect_via(app)
def on_appctx_down(sender, exc=None, **extra):
    events.append("appcontext_tearing_down")


appcontext_pushed.connect(lambda sender, **kw: events.append("appcontext_pushed"), app, weak=False)
appcontext_popped.connect(lambda sender, **kw: events.append("appcontext_popped"), app, weak=False)

reply = app.test_client().get("/books")
assert reply.status_code == 200
assert events == [
    "appcontext_pushed",
    "request_started GET /books",
    "request_finished 200",
    "request_tearing_down exc=None",
    "appcontext_tearing_down",
    "appcontext_popped",
]

from blinker import Namespace

demo = Namespace()
pinged = demo.signal("pinged")
calls: list[str] = []


def register_weak():
    def receiver(sender, **extra):
        calls.append("weak receiver")
    pinged.connect(receiver, weak=True)       # only the signal knows it, through a weakref


def register_strong():
    def receiver(sender, **extra):
        calls.append("strong receiver")
    pinged.connect(receiver, weak=False)      # the signal keeps the function alive


register_weak()
register_strong()
import gc
gc.collect()

pinged.send("bookshelf")
assert calls == ["strong receiver"]           # the weak one vanished with its scope

# connect_via and connected_to: lifetime and sender filtering
seen = []


@pinged.connect_via("shop")
def only_for_shop(sender, **extra):
    seen.append(("via", sender))


with pinged.connected_to(lambda sender, **kw: seen.append(("with", sender)), "library"):
    pinged.send("shop")
    pinged.send("library")
pinged.send("library")                        # the connected_to receiver is gone now

assert seen == [("via", "shop"), ("with", "library")]
assert pinged.has_receivers_for("shop") and pinged.has_receivers_for("library")

from flask import request_started

failing = Flask("failing")
failing.config.update(SECRET_KEY=secrets.token_hex())


@failing.get("/")
def home():
    return "ok"


def explode(sender, **extra):
    raise RuntimeError("receiver bug")


with request_started.connected_to(explode, failing):
    assert failing.test_client().get("/").status_code == 500   # the signal broke the request

assert failing.test_client().get("/").status_code == 200       # connected_to has disconnected it

from blinker import Namespace
from flask import current_app

bookshelf_signals = Namespace()
book_added = bookshelf_signals.signal("book-added", doc="Sent after a book was stored. Arguments: book.")
book_removed = bookshelf_signals.signal("book-removed")

assert book_added.name == "book-added"
assert bookshelf_signals.signal("book-added") is book_added   # same name, same object

shelf = Flask("shelf")
shelf.config.update(SECRET_KEY=secrets.token_hex())
audit_log: list[str] = []
cache = {"all-books": ["stale"]}
catalog: dict[int, dict] = {}


def add_book(title: str) -> dict:
    book = {"id": len(catalog) + 1, "title": title}
    catalog[book["id"]] = book
    # The real application object is the sender, never the current_app proxy.
    results = book_added.send(current_app._get_current_object(), book=book)
    return {"book": book, "receivers": len(results)}


@book_added.connect_via(shelf)
def audit(sender, book, **extra):
    audit_log.append(f"added {book['id']}: {book['title']}")


@book_added.connect_via(shelf)
def invalidate_cache(sender, book, **extra):
    cache.pop("all-books", None)


@shelf.post("/books/<title>")
def create_book(title):
    return add_book(title), 201


reply = shelf.test_client().post("/books/Dune")
assert reply.status_code == 201 and reply.get_json()["receivers"] == 2
assert audit_log == ["added 1: Dune"] and "all-books" not in cache

# A signal without receivers costs almost nothing; skip expensive payloads with has_receivers_for.
assert book_removed.has_receivers_for(shelf) is False

unhandled = Flask("unhandled")
unhandled.config.update(SECRET_KEY=secrets.token_hex(), PROPAGATE_EXCEPTIONS=False)
trace: list[str] = []


@unhandled.get("/boom")
def boom():
    raise ValueError("unexpected")


@unhandled.get("/gone")
def gone():
    from flask import abort
    abort(404)


from flask import got_request_exception, request_tearing_down

got_request_exception.connect(lambda s, exception, **kw: trace.append(f"exception {type(exception).__name__}"), unhandled, weak=False)
request_tearing_down.connect(lambda s, exc=None, **kw: trace.append(f"teardown {type(exc).__name__}"), unhandled, weak=False)

client = unhandled.test_client()
assert client.get("/boom").status_code == 500
assert client.get("/gone").status_code == 404
assert trace == ["exception ValueError", "teardown ValueError", "teardown NoneType"]

import asyncio

from flask import message_flashed

asyncio_app = Flask("asyncio_app")
asyncio_app.config.update(SECRET_KEY=secrets.token_hex())
flashed: list[tuple[str, str]] = []


@asyncio_app.get("/flash")
def do_flash():
    flash("Saved", "success")
    return "done"


@message_flashed.connect_via(asyncio_app)
async def record_flash(sender, message, category, **extra):
    await asyncio.sleep(0)                       # any await works inside the receiver
    flashed.append((category, message))


assert asyncio_app.test_client().get("/flash").status_code == 200
assert flashed == [("success", "Saved")]

# A custom signal: send() needs a wrapper for coroutines, send_async() awaits them directly.
price_changed = bookshelf_signals.signal("price-changed")
log: list[str] = []


@price_changed.connect_via("shop")
async def notify(sender, title, **extra):
    log.append(f"async {title}")


@price_changed.connect_via("shop")
def count(sender, title, **extra):
    log.append(f"sync {title}")


try:
    price_changed.send("shop", title="Dune")
except RuntimeError as error:
    assert "coroutine" in str(error)             # no _async_wrapper, so the coroutine receiver is refused

log.clear()


def run_in_thread(func):
    async def wrapper(*args, **kwargs):
        return await asyncio.to_thread(func, *args, **kwargs)
    return wrapper


results = asyncio.run(price_changed.send_async("shop", _sync_wrapper=run_in_thread, title="Dune"))
assert len(results) == 2 and sorted(log) == ["async Dune", "sync Dune"]

from contextlib import contextmanager

from flask import template_rendered

pages = Flask("pages")
pages.config.update(SECRET_KEY=secrets.token_hex())


@pages.get("/")
def index():
    flash("Welcome", "info")
    return render_template_string("{% for b in books %}<li>{{ b }}</li>{% endfor %}", books=["Dune", "Emma"])


@contextmanager
def captured_templates(app):
    recorded = []

    def record(sender, template, context, **extra):
        recorded.append((template, context))

    with template_rendered.connected_to(record, app):
        yield recorded


@contextmanager
def captured_flashes(app):
    recorded = []

    def record(sender, message, category, **extra):
        recorded.append((category, message))

    with message_flashed.connected_to(record, app):
        yield recorded


with captured_templates(pages) as templates, captured_flashes(pages) as flashes:
    reply = pages.test_client().get("/")

assert reply.status_code == 200
assert len(templates) == 1
template, context = templates[0]
assert template.name is None                      # a string template has no name; file templates do
assert context["books"] == ["Dune", "Emma"]
assert flashes == [("info", "Welcome")]