server.Redirect: add status param to constructor

README.md

@@ -227,6 +227,10 @@ Here's how to package, test, and ship a new release.
 
 ## Changelog
 
+### 0.8 - unreleased
+* `server`:
+  * Add `status` param to `Redirect`.
+
 ### 0.7 - 2024-06-24
 
 * Fix websocket subscription server hang with blocking server XRPC methods due to exhausting worker thread pool ([#8](https://github.com/snarfed/lexrpc/issues/8)).

lexrpc/flask_server.py

@@ -93,7 +93,7 @@ def dispatch_request(self, nsid):
             # io.BufferedReader/Writer?
             output = self.server.call(nsid, input=input, **params)
         except Redirect as r:
-            return redirect(r.to)
+            return redirect(r.to, code=r.status)
         except NotImplementedError as e:
             return {
                 'error': 'MethodNotImplemented',

lexrpc/server.py

@@ -10,17 +10,18 @@
 class Redirect(Exception):
     """Raised by XRPC handlers to direct the server to serve an HTTP redirect.
 
-    Uses the HTTP 302 status code.
-
     Whether this is official supported by the XRPC spec is still TBD:
     https://github.com/bluesky-social/atproto/discussions/1228
 
     Attributes:
       to (str): URL to redirect to
+      status (int): HTTP status code, defaults to 302
     """
-    def __init__(self, to):
+    def __init__(self, to, status=302):
         assert to
+        assert status
         self.to = to
+        self.status = status
 
 
 class Server(Base):

lexrpc/tests/test_flask_server.py

@@ -265,9 +265,16 @@ def test_encodings(self):
 
     def test_redirect(self):
         @server.method('io.example.redirect')
-        def redirect(input):
-            raise Redirect('http://to/here')
+        def redirect(input, status=None):
+            kwargs = {}
+            if status:
+                kwargs['status'] = status
+            raise Redirect('http://to/here', **kwargs)
 
         resp = self.client.post('/xrpc/io.example.redirect')
         self.assertEqual(302, resp.status_code)
         self.assertEqual('http://to/here', resp.headers['Location'])
+
+        resp = self.client.post('/xrpc/io.example.redirect?status=301')
+        self.assertEqual(301, resp.status_code)
+        self.assertEqual('http://to/here', resp.headers['Location'])