如何从Flask转发请求至其他API应用及创建动态请求转发处理器

Got it, let's tackle your two Flask-related requirements one by one—here's a practical breakdown with code examples that you can adapt for your use case:

The most straightforward way to handle this is using the requests library to relay incoming Flask requests to your target API. You'll want to preserve key details from the original request (like HTTP method, headers, query params, and request body) to ensure the target API receives the correct data.

Step-by-Step Implementation: #

• First, install the required package if you haven't already:

  pip install requests flask
  

• Create a reusable forwarding function that handles all request types:

  from flask import Flask, request, Response
  import requests
  
  app = Flask(__name__)
  
  def forward_request(target_url):
      # Copy original request headers, remove Flask-specific ones that might cause issues
      headers = dict(request.headers)
      headers.pop('Host', None)
  
      # Relay the full request details to the target API
      response = requests.request(
          method=request.method,
          url=target_url,
          headers=headers,
          data=request.get_data(),
          params=request.args,
          allow_redirects=False
      )
  
      # Return the target API's response to the original client, excluding headers Flask manages internally
      excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
      response_headers = [(name, value) for name, value in response.raw.headers.items()
                          if name.lower() not in excluded_headers]
      return Response(response.content, response.status_code, response_headers)
  
  # Example route that forwards to a specific API endpoint
  @app.route('/forward-to-api', methods=['GET', 'POST', 'PUT', 'DELETE'])
  def forward_to_api():
      target_api_url = "https://your-target-api.com/desired-endpoint"
      return forward_request(target_api_url)
  
  if __name__ == '__main__':
      app.run(debug=True)
  

This function takes care of relaying all critical request data and returns the target API's response exactly as received (minus headers that Flask handles internally).

To build a processor that routes requests based on a database-backed mapping, you'll need three core components: a database table to store route-to-target mappings, a catch-all Flask route, and logic to look up the target API and forward the request.

Step 1: Database Setup (Using SQLAlchemy) #

First, define a model to store your route mappings:

from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///route_mappings.db'  # Replace with your actual DB URI
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

class RouteMapping(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    endpoint_path = db.Column(db.String(255), unique=True, nullable=False)  # e.g., "/users/*"
    target_api_url = db.Column(db.String(255), nullable=False)  # e.g., "https://user-service.com/"
    allowed_methods = db.Column(db.String(100), nullable=False)  # e.g., "GET,POST,PUT"

# Create the database tables (run this once)
with app.app_context():
    db.create_all()

Step 2: Dynamic Routing Logic #

Add a catch-all route that looks up the mapping, validates the request method, and forwards the request:

from functools import lru_cache

# Optional: Cache mappings to avoid frequent DB hits (refresh every 5 minutes or on demand)
@lru_cache(maxsize=128)
def get_route_mappings():
    mappings = RouteMapping.query.all()
    return {mapping.endpoint_path: mapping for mapping in mappings}

@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
def dynamic_router(path):
    full_request_path = f"/{path}"
    mappings = get_route_mappings()

    # Find the matching route mapping (adjust this to support wildcards if needed)
    target_mapping = None
    for endpoint_pattern, mapping in mappings.items():
        if full_request_path.startswith(endpoint_pattern):
            target_mapping = mapping
            break

    if not target_mapping:
        return "Route not found", 404

    # Validate the request method is allowed for this route
    allowed_methods = target_mapping.allowed_methods.split(',')
    if request.method not in allowed_methods:
        return "Method not allowed", 405

    # Build the full target URL (append the remaining path from the request)
    target_url = f"{target_mapping.target_api_url}{full_request_path[len(target_mapping.endpoint_path):]}"

    # Forward the request using our reusable function
    return forward_request(target_url)

# Optional: Endpoint to manually refresh the route cache when mappings are updated
@app.route('/refresh-routes', methods=['POST'])
def refresh_routes():
    get_route_mappings.cache_clear()
    return "Route cache refreshed successfully", 200

Quick Notes: #

• For wildcard support (like /users/*), replace the startswith check with regex matching to fit your pattern needs.
• The lru_cache reduces database load—call the /refresh-routes endpoint whenever you update route mappings in the database.
• Add try/except blocks around the requests call to handle target API timeouts or errors gracefully.

内容的提问来源于stack exchange，提问作者angga prabuwisesa