rorycl · July 31, 2025 14:12
diff --git a/flask_endpoint_example.py b/flask_endpoint_example.py
 """
 Defines the Flask endpoints for an example search and user list API.

 This module contains the view functions that handle HTTP requests for searching
 documents and managing user-specific lists. It follows the Application Factory
 pattern, where endpoints are registered on a Blueprint, which is then attached
 to the main Flask app instance.

 Responsibilities:
 - Defining URL routes and accepted HTTP methods.
 - Handling request authentication and authorization via decorators.
 - Validating and parsing incoming request data (query params, JSON bodies).
 - Orchestrating calls to the business logic layer (services).
 - Handling exceptions from lower layers and transforming them into appropriate
  HTTP responses.
 - Rendering HTML templates or returning JSON responses based on content
  negotiation.
 """

 import logging

 from flask import (
    Blueprint,
    Response,
    g,
    jsonify,
    render_template,
    request,
    current_app,
 )
 from pydantic import ValidationError

 # Made up local imports for services and validation schemas
 from .auth import session_handler
 from .schemas import DocumentSearchSchema, ListItemAddSchema
 from .services import (
    DatabaseService,
    SearchService,
    db_exceptions,
    search_exceptions,
 )

 # A flask blueprint is a set of operations which makes the app more modular.
 bp = Blueprint("main", __name__)

 # Configure logging for this module (__name__ automatically works for sub-modules).
 log = logging.getLogger(__name__)


 def get_search_service() -> SearchService:
    """
    Get the application's search service instance, if needed.
    """
    if "search_service" not in g:
        g.search_service = SearchService(dsn=current_app.config["SEARCHER_DSN"])
    return g.search_service


 def get_db_service() -> DatabaseService:
    """
    Get the application's database service, if needed.
    """
    if "db_service" not in g:
        g.db_service = DatabaseService(dsn=current_app.config["DB_DSN"])
    return g.db_service


 @app.teardown_appcontext
 def teardown_connections(exception=None):
    """
    Close the connections at the end of the request
    """
    db = g.pop("db_service", None)
    if db is not None:
        db.close()

    searcher = g.pop("search_service", None)
    if searcher is not None:
        searcher.close()


 @bp.route("/search", methods=["GET"])
 @session_handler(guest_ok=True)
 def document_search() -> Response:
    """
    Searches documents based on query parameters.

    This endpoint is publicly accessible to both guests and authenticated users.
    It validates search parameters, performs a search using the SearchService,
    and renders an HTML results page.

    Query Params:
        terms (str): The search terms. Required, min 3 chars.
        page (in): The page number for pagination. Default 1.

    Returns:
        A rendered HTML page with search results or an error message.
    """

    # validate
    try:
        search_params = DocumentSearchSchema.model_validate(request.args.to_dict())
    except ValidationError as e:
        log.info(f"Invalid search request: {e.errors()}")
        return (
            render_template(
                "search_results.html",
                validation_errors=e.errors(),
                search_params=request.args,
            ),
            422,
        )

    # search
    try:
        search_service = get_search_service()
        search_results = search_service.find_documents(
            user=g.user, query=search_params.terms, page=search_params.page
        )
        return render_template(
            "search_results.html",
            results=search_results,
            search_params=search_params,
        )

    except search_exceptions.NoResultsFound:
        log.info(f"No results found for query: '{search_params.terms}'")
        return render_template(
            "search_results.html",
            results=None,
            search_params=search_params,
        )

    except search_exceptions.SearchServiceError as e:
        log.exception(
            f"Search service failed for query: '{search_params.terms}'. User {g.user.id}"
        )
        return render_template("errors/500.html"), 500


 @bp.route("/lists/add-item", methods=["POST"])
 @session_handler(guest_ok=False)  # This endpoint requires a logged-in user
 def add_items_to_list() -> Response:
    """
    Adds a document to a user's list.

    This endpoint is for authenticated users only. It accepts a JSON body
    specifying the document and list. It supports content negotiation to
    return either a JSON response (for JS clients) or an HTML redirect (for
    form submissions).

    JSON Body:
        doc_id (str): The unique document identifier.
        list_id (str, optional): The user's list identifier. Defaults to the
                                 user's "favourites" list.

    Returns:
        A JSON object with the updated list contents or an HTML redirect,
        depending on the request's 'Accept' header.
    """
    if not request.is_json:
        return jsonify({"error": "Request must be JSON"}), 415  # Unsupported Media Type

    # validate
    try:
        item_data = ListItemAddSchema.model_validate(request.json)
    except ValidationError as e:
        log.info(f"Invalid list item data from user {g.user.id}: {e.errors()}")

    # add document to list
    try:
        db_service = get_db_service()
        updated_list = db_service.add_document_to_list(
            user=g.user, doc_id=item_data.doc_id, list_id=item_data.list_id
        )
        if (
            request.accept_mimetypes.accept_json
            and not request.accept_mimetypes.accept_html
        ):
            return jsonify(updated_list.model_dump()), 200
        else:
            return redirect(url_for("main.show_list", list_id=updated_list.id))

    except db_exceptions.PermissionDeniedError:
        log.warning(
            f"Permission denied: user {g.uder.ud} tried to add doc {item_data.doc_id} to a list."
        )
        return jsonify({"error": "Permission denied to access this document."}), 403

    except db_exceptions.DocumentNotFoundError:
        log.info(f"User {g.user.id} tried to add non-existent doc {item_data.doc_id}.")
        return jsonify(
            {"error": f"Document with ID {item_data.doc_id} not found."}
        ), 404

    except db_exceptions.DatabaseServiceError as e:
        log.exception(
            f"Database service failed while user {g.user.id} added doc {item_data.doc_id}."
        )
        return jsonify({"error": "An internal server error occurred."}), 500


 if __name__ == '__main__':
    # from . import views

    app = Flask(__name__)
    # ... other config ...

    # Register the blueprint
    app.register_blueprint(views.bp)

    # Register the teardown function
    app.teardown_appcontext(views.teardown_connections)

    return app
	"""
	Defines the Flask endpoints for an example search and user list API.

	This module contains the view functions that handle HTTP requests for searching
	documents and managing user-specific lists. It follows the Application Factory
	pattern, where endpoints are registered on a Blueprint, which is then attached
	to the main Flask app instance.

	Responsibilities:
	- Defining URL routes and accepted HTTP methods.
	- Handling request authentication and authorization via decorators.
	- Validating and parsing incoming request data (query params, JSON bodies).
	- Orchestrating calls to the business logic layer (services).
	- Handling exceptions from lower layers and transforming them into appropriate
	HTTP responses.
	- Rendering HTML templates or returning JSON responses based on content
	negotiation.
	"""

	import logging

	from flask import (
	Blueprint,
	Response,
	g,
	jsonify,
	render_template,
	request,
	current_app,
	)
	from pydantic import ValidationError

	# Made up local imports for services and validation schemas
	from .auth import session_handler
	from .schemas import DocumentSearchSchema, ListItemAddSchema
	from .services import (
	DatabaseService,
	SearchService,
	db_exceptions,
	search_exceptions,
	)

	# A flask blueprint is a set of operations which makes the app more modular.
	bp = Blueprint("main", __name__)

	# Configure logging for this module (__name__ automatically works for sub-modules).
	log = logging.getLogger(__name__)


	def get_search_service() -> SearchService:
	"""
	Get the application's search service instance, if needed.
	"""
	if "search_service" not in g:
	g.search_service = SearchService(dsn=current_app.config["SEARCHER_DSN"])
	return g.search_service


	def get_db_service() -> DatabaseService:
	"""
	Get the application's database service, if needed.
	"""
	if "db_service" not in g:
	g.db_service = DatabaseService(dsn=current_app.config["DB_DSN"])
	return g.db_service


	@app.teardown_appcontext
	def teardown_connections(exception=None):
	"""
	Close the connections at the end of the request
	"""
	db = g.pop("db_service", None)
	if db is not None:
	db.close()

	searcher = g.pop("search_service", None)
	if searcher is not None:
	searcher.close()


	@bp.route("/search", methods=["GET"])
	@session_handler(guest_ok=True)
	def document_search() -> Response:
	"""
	Searches documents based on query parameters.

	This endpoint is publicly accessible to both guests and authenticated users.
	It validates search parameters, performs a search using the SearchService,
	and renders an HTML results page.

	Query Params:
	terms (str): The search terms. Required, min 3 chars.
	page (in): The page number for pagination. Default 1.

	Returns:
	A rendered HTML page with search results or an error message.
	"""

	# validate
	try:
	search_params = DocumentSearchSchema.model_validate(request.args.to_dict())
	except ValidationError as e:
	log.info(f"Invalid search request: {e.errors()}")
	return (
	render_template(
	"search_results.html",
	validation_errors=e.errors(),
	search_params=request.args,
	),
	422,
	)

	# search
	try:
	search_service = get_search_service()
	search_results = search_service.find_documents(
	user=g.user, query=search_params.terms, page=search_params.page
	)
	return render_template(
	"search_results.html",
	results=search_results,
	search_params=search_params,
	)

	except search_exceptions.NoResultsFound:
	log.info(f"No results found for query: '{search_params.terms}'")
	return render_template(
	"search_results.html",
	results=None,
	search_params=search_params,
	)

	except search_exceptions.SearchServiceError as e:
	log.exception(
	f"Search service failed for query: '{search_params.terms}'. User {g.user.id}"
	)
	return render_template("errors/500.html"), 500


	@bp.route("/lists/add-item", methods=["POST"])
	@session_handler(guest_ok=False) # This endpoint requires a logged-in user
	def add_items_to_list() -> Response:
	"""
	Adds a document to a user's list.

	This endpoint is for authenticated users only. It accepts a JSON body
	specifying the document and list. It supports content negotiation to
	return either a JSON response (for JS clients) or an HTML redirect (for
	form submissions).

	JSON Body:
	doc_id (str): The unique document identifier.
	list_id (str, optional): The user's list identifier. Defaults to the
	user's "favourites" list.

	Returns:
	A JSON object with the updated list contents or an HTML redirect,
	depending on the request's 'Accept' header.
	"""
	if not request.is_json:
	return jsonify({"error": "Request must be JSON"}), 415 # Unsupported Media Type

	# validate
	try:
	item_data = ListItemAddSchema.model_validate(request.json)
	except ValidationError as e:
	log.info(f"Invalid list item data from user {g.user.id}: {e.errors()}")

	# add document to list
	try:
	db_service = get_db_service()
	updated_list = db_service.add_document_to_list(
	user=g.user, doc_id=item_data.doc_id, list_id=item_data.list_id
	)
	if (
	request.accept_mimetypes.accept_json
	and not request.accept_mimetypes.accept_html
	):
	return jsonify(updated_list.model_dump()), 200
	else:
	return redirect(url_for("main.show_list", list_id=updated_list.id))

	except db_exceptions.PermissionDeniedError:
	log.warning(
	f"Permission denied: user {g.uder.ud} tried to add doc {item_data.doc_id} to a list."
	)
	return jsonify({"error": "Permission denied to access this document."}), 403

	except db_exceptions.DocumentNotFoundError:
	log.info(f"User {g.user.id} tried to add non-existent doc {item_data.doc_id}.")
	return jsonify(
	{"error": f"Document with ID {item_data.doc_id} not found."}
	), 404

	except db_exceptions.DatabaseServiceError as e:
	log.exception(
	f"Database service failed while user {g.user.id} added doc {item_data.doc_id}."
	)
	return jsonify({"error": "An internal server error occurred."}), 500


	if __name__ == '__main__':
	# from . import views

	app = Flask(__name__)
	# ... other config ...

	# Register the blueprint
	app.register_blueprint(views.bp)

	# Register the teardown function
	app.teardown_appcontext(views.teardown_connections)

	return app
No results found