Introduce builder for C++ Flight servers

cpp/src/arrow/flight/client.cc

@@ -231,17 +231,20 @@ class DoPutPayloadWriter : public ipc::internal::IpcPayloadWriter {
 
 class FlightClient::FlightClientImpl {
  public:
-  Status Connect(const Location& location) {
+  Status Connect(const Location& location, const FlightClientOptions& options) {
     const std::string& scheme = location.scheme();
 
     std::stringstream grpc_uri;
     std::shared_ptr<grpc::ChannelCredentials> creds;
     if (scheme == kSchemeGrpc || scheme == kSchemeGrpcTcp || scheme == kSchemeGrpcTls) {
-      // TODO(wesm): Support other kinds of GRPC ChannelCredentials
       grpc_uri << location.uri_->host() << ":" << location.uri_->port_text();
 
       if (scheme == "grpc+tls") {
-        creds = grpc::SslCredentials(grpc::SslCredentialsOptions());
+        grpc::SslCredentialsOptions ssl_options;
+        if (!options.tls_root_certs.empty()) {
+          ssl_options.pem_root_certs = options.tls_root_certs;
+        }
+        creds = grpc::SslCredentials(ssl_options);
       } else {
         creds = grpc::InsecureChannelCredentials();
       }
@@ -401,8 +404,13 @@ FlightClient::~FlightClient() {}
 
 Status FlightClient::Connect(const Location& location,
                              std::unique_ptr<FlightClient>* client) {
+  return Connect(location, {}, client);
+}
+
+Status FlightClient::Connect(const Location& location, const FlightClientOptions& options,
+                             std::unique_ptr<FlightClient>* client) {
   client->reset(new FlightClient);
-  return (*client)->impl_->Connect(location);
+  return (*client)->impl_->Connect(location, options);
 }
 
 Status FlightClient::Authenticate(const FlightCallOptions& options,

cpp/src/arrow/flight/client.h

@@ -57,6 +57,11 @@ class ARROW_EXPORT FlightCallOptions {
   TimeoutDuration timeout;
 };
 
+class ARROW_EXPORT FlightClientOptions {
+ public:
+  std::string tls_root_certs;
+};
+
 /// \brief Client class for Arrow Flight RPC services (gRPC-based).
 /// API experimental for now
 class ARROW_EXPORT FlightClient {
@@ -70,6 +75,15 @@ class ARROW_EXPORT FlightClient {
   /// successful
   static Status Connect(const Location& location, std::unique_ptr<FlightClient>* client);
 
+  /// \brief Connect to an unauthenticated flight service
+  /// \param[in] location the URI
+  /// \param[in] options Other options for setting up the client
+  /// \param[out] client the created FlightClient
+  /// \return Status OK status may not indicate that the connection was
+  /// successful
+  static Status Connect(const Location& location, const FlightClientOptions& options,
+                        std::unique_ptr<FlightClient>* client);
+
   /// \brief Authenticate to the server using the given handler.
   /// \param[in] options Per-RPC options
   /// \param[in] auth_handler The authentication mechanism to use

cpp/src/arrow/flight/flight-test.cc

@@ -255,24 +255,25 @@ class DoPutTestServer : public FlightServerBase {
 class TestAuthHandler : public ::testing::Test {
  public:
   void SetUp() {
-    port_ = 30000;
-    server_.reset(new InProcessTestServer(
-        std::unique_ptr<FlightServerBase>(new AuthTestServer), port_));
-    ASSERT_OK(server_->Start(std::unique_ptr<ServerAuthHandler>(
-        new TestServerAuthHandler("user", "p4ssw0rd"))));
+    Location location;
+    std::unique_ptr<FlightServerBase> server(new AuthTestServer);
+
+    ASSERT_OK(Location::ForGrpcTcp("localhost", 30000, &location));
+    FlightServerOptions options(location);
+    options.auth_handler =
+        std::unique_ptr<ServerAuthHandler>(new TestServerAuthHandler("user", "p4ssw0rd"));
+    ASSERT_OK(server->Init(options));
+
+    server_.reset(new InProcessTestServer(std::move(server), location));
+    ASSERT_OK(server_->Start());
     ASSERT_OK(ConnectClient());
   }
 
   void TearDown() { server_->Stop(); }
 
-  Status ConnectClient() {
-    Location location;
-    RETURN_NOT_OK(Location::ForGrpcTcp("localhost", port_, &location));
-    return FlightClient::Connect(location, &client_);
-  }
+  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
 
  protected:
-  int port_;
   std::unique_ptr<FlightClient> client_;
   std::unique_ptr<InProcessTestServer> server_;
 };
@@ -280,17 +281,21 @@ class TestAuthHandler : public ::testing::Test {
 class TestDoPut : public ::testing::Test {
  public:
   void SetUp() {
-    port_ = 30000;
+    Location location;
+    ASSERT_OK(Location::ForGrpcTcp("localhost", 30000, &location));
+
     do_put_server_ = new DoPutTestServer();
     server_.reset(new InProcessTestServer(
-        std::unique_ptr<FlightServerBase>(do_put_server_), port_));
-    ASSERT_OK(server_->Start({}));
+        std::unique_ptr<FlightServerBase>(do_put_server_), location));
+    FlightServerOptions options(location);
+    ASSERT_OK(do_put_server_->Init(options));
+    ASSERT_OK(server_->Start());
     ASSERT_OK(ConnectClient());
   }
 
   void TearDown() { server_->Stop(); }
 
-  Status ConnectClient() { return FlightClient::Connect("localhost", port_, &client_); }
+  Status ConnectClient() { return FlightClient::Connect(server_->location(), &client_); }
 
   void CheckBatches(FlightDescriptor expected_descriptor,
                     const BatchVector& expected_batches) {
@@ -314,7 +319,6 @@ class TestDoPut : public ::testing::Test {
   }
 
  protected:
-  int port_;
   std::unique_ptr<FlightClient> client_;
   std::unique_ptr<InProcessTestServer> server_;
   DoPutTestServer* do_put_server_;

cpp/src/arrow/flight/perf-server.cc

@@ -207,8 +207,9 @@ int main(int argc, char** argv) {
 
   arrow::flight::Location location;
   ARROW_CHECK_OK(arrow::flight::Location::ForGrpcTcp("0.0.0.0", FLAGS_port, &location));
-  ARROW_CHECK_OK(
-      g_server->Init(std::unique_ptr<arrow::flight::NoOpAuthHandler>(), location));
+  arrow::flight::FlightServerOptions options(location);
+
+  ARROW_CHECK_OK(g_server->Init(options));
   // Exit with a clean error code (0) on SIGTERM
   ARROW_CHECK_OK(g_server->SetShutdownOnSignals({SIGTERM}));
   std::cout << "Server port: " << FLAGS_port << std::endl;

cpp/src/arrow/flight/server.cc

@@ -21,6 +21,7 @@
 #include <atomic>
 #include <cstdint>
 #include <memory>
+#include <sstream>
 #include <string>
 #include <utility>
 
@@ -438,24 +439,38 @@ void FlightServerBase::Impl::HandleSignal(int signum) {
   }
 }
 
+FlightServerOptions::FlightServerOptions(const Location& location_)
+    : location(location_), auth_handler(nullptr) {}
+
 FlightServerBase::FlightServerBase() { impl_.reset(new Impl); }
 
 FlightServerBase::~FlightServerBase() {}
 
-Status FlightServerBase::Init(std::unique_ptr<ServerAuthHandler> auth_handler,
-                              const Location& location) {
-  std::shared_ptr<ServerAuthHandler> handler = std::move(auth_handler);
+Status FlightServerBase::Init(FlightServerOptions& options) {
+  std::shared_ptr<ServerAuthHandler> handler = std::move(options.auth_handler);
   impl_->service_.reset(new FlightServiceImpl(handler, this));
 
   grpc::ServerBuilder builder;
   // Allow uploading messages of any length
   builder.SetMaxReceiveMessageSize(-1);
 
+  const Location& location = options.location;
   const std::string scheme = location.scheme();
-  if (scheme == kSchemeGrpc || scheme == kSchemeGrpcTcp) {
+  if (scheme == kSchemeGrpc || scheme == kSchemeGrpcTcp || scheme == kSchemeGrpcTls) {
     std::stringstream address;
     address << location.uri_->host() << ':' << location.uri_->port_text();
-    builder.AddListeningPort(address.str(), grpc::InsecureServerCredentials());
+
+    std::shared_ptr<grpc::ServerCredentials> creds;
+    if (scheme == kSchemeGrpcTls) {
+      grpc::SslServerCredentialsOptions ssl_options;
+      ssl_options.pem_key_cert_pairs.push_back(
+          {options.tls_private_key, options.tls_cert_chain});
+      creds = grpc::SslServerCredentials(ssl_options);
+    } else {
+      creds = grpc::InsecureServerCredentials();
+    }
+
+    builder.AddListeningPort(address.str(), creds);
   } else if (scheme == kSchemeGrpcUnix) {
     std::stringstream address;
     address << "unix:" << location.uri_->path();

cpp/src/arrow/flight/server.h

@@ -24,6 +24,7 @@
 #include <string>
 #include <vector>
 
+#include "arrow/flight/server_auth.h"
 #include "arrow/flight/types.h"  // IWYU pragma: keep
 #include "arrow/ipc/dictionary.h"
 #include "arrow/memory_pool.h"
@@ -38,8 +39,6 @@ class Status;
 
 namespace flight {
 
-class ServerAuthHandler;
-
 /// \brief Interface that produces a sequence of IPC payloads to be sent in
 /// FlightData protobuf messages
 class ARROW_EXPORT FlightDataStream {
@@ -89,6 +88,16 @@ class ARROW_EXPORT ServerCallContext {
   virtual const std::string& peer_identity() const = 0;
 };
 
+class ARROW_EXPORT FlightServerOptions {
+ public:
+  explicit FlightServerOptions(const Location& location_);
+
+  Location location;
+  std::unique_ptr<ServerAuthHandler> auth_handler;
+  std::string tls_cert_chain;
+  std::string tls_private_key;
+};
+
 /// \brief Skeleton RPC server implementation which can be used to create
 /// custom servers by implementing its abstract methods
 class ARROW_EXPORT FlightServerBase {
@@ -100,10 +109,8 @@ class ARROW_EXPORT FlightServerBase {
 
   /// \brief Initialize a Flight server listening at the given location.
   /// This method must be called before any other method.
-  /// \param[in] auth_handler The authentication handler. May be
-  /// nullptr if no authentication is desired.
-  /// \param[in] location The location to serve on.
-  Status Init(std::unique_ptr<ServerAuthHandler> auth_handler, const Location& location);
+  /// \param[in] options The configuration for this server.
+  Status Init(FlightServerOptions& options);
 
   /// \brief Set the server to stop when receiving any of the given signal
   /// numbers.

cpp/src/arrow/flight/test-integration-server.cc

@@ -126,7 +126,9 @@ int main(int argc, char** argv) {
   g_server.reset(new arrow::flight::FlightIntegrationTestServer);
   arrow::flight::Location location;
   ARROW_CHECK_OK(arrow::flight::Location::ForGrpcTcp("0.0.0.0", FLAGS_port, &location));
-  ARROW_CHECK_OK(g_server->Init(nullptr, location));
+  arrow::flight::FlightServerOptions options(location);
+
+  ARROW_CHECK_OK(g_server->Init(options));
   // Exit with a clean error code (0) on SIGTERM
   ARROW_CHECK_OK(g_server->SetShutdownOnSignals({SIGTERM}));
 

cpp/src/arrow/flight/test-server.cc

@@ -139,10 +139,12 @@ int main(int argc, char** argv) {
   gflags::ParseCommandLineFlags(&argc, &argv, true);
 
   g_server.reset(new arrow::flight::FlightTestServer);
+
   arrow::flight::Location location;
   ARROW_CHECK_OK(arrow::flight::Location::ForGrpcTcp("0.0.0.0", FLAGS_port, &location));
-  ARROW_CHECK_OK(
-      g_server->Init(std::unique_ptr<arrow::flight::NoOpAuthHandler>(), location));
+  arrow::flight::FlightServerOptions options(location);
+
+  ARROW_CHECK_OK(g_server->Init(options));
   // Exit with a clean error code (0) on SIGTERM
   ARROW_CHECK_OK(g_server->SetShutdownOnSignals({SIGTERM}));
 

cpp/src/arrow/flight/test-util.cc

@@ -118,22 +118,17 @@ bool TestServer::IsRunning() { return server_process_->running(); }
 
 int TestServer::port() const { return port_; }
 
-Status InProcessTestServer::Start(std::unique_ptr<ServerAuthHandler> auth_handler) {
-  Location location;
-  RETURN_NOT_OK(Location::ForGrpcTcp("localhost", port_, &location));
-  RETURN_NOT_OK(server_->Init(std::move(auth_handler), location));
+Status InProcessTestServer::Start() {
   thread_ = std::thread([this]() { ARROW_EXPECT_OK(server_->Serve()); });
   return Status::OK();
 }
 
-Status InProcessTestServer::Start() { return Start({}); }
-
 void InProcessTestServer::Stop() {
   server_->Shutdown();
   thread_.join();
 }
 
-int InProcessTestServer::port() const { return port_; }
+const Location& InProcessTestServer::location() const { return location_; }
 
 InProcessTestServer::~InProcessTestServer() {
   // Make sure server shuts down properly

cpp/src/arrow/flight/test-util.h

@@ -63,17 +63,17 @@ class ARROW_EXPORT TestServer {
 
 class ARROW_EXPORT InProcessTestServer {
  public:
-  explicit InProcessTestServer(std::unique_ptr<FlightServerBase> server, int port)
-      : server_(std::move(server)), port_(port), thread_() {}
+  explicit InProcessTestServer(std::unique_ptr<FlightServerBase> server,
+                               const Location& location)
+      : server_(std::move(server)), location_(location), thread_() {}
   ~InProcessTestServer();
   Status Start();
-  Status Start(std::unique_ptr<ServerAuthHandler> auth_handler);
   void Stop();
-  int port() const;
+  const Location& location() const;
 
  private:
   std::unique_ptr<FlightServerBase> server_;
-  int port_;
+  Location location_;
   std::thread thread_;
 };
 

cpp/src/arrow/flight/types.cc

@@ -102,7 +102,7 @@ Status Location::ForGrpcUnix(const std::string& path, Location* location) {
   return Location::Parse(uri_string.str(), location);
 }
 
-std::string Location::ToString() const { return uri_->to_string(); }
+std::string Location::ToString() const { return uri_->ToString(); }
 std::string Location::scheme() const {
   std::string scheme = uri_->scheme();
   if (scheme.empty()) {

cpp/src/arrow/util/uri.cc

@@ -120,7 +120,7 @@ std::string Uri::path() const {
   return ss.str();
 }
 
-const std::string& Uri::to_string() const { return impl_->string_rep_; }
+const std::string& Uri::ToString() const { return impl_->string_rep_; }
 
 Status Uri::Parse(const std::string& uri_string) {
   impl_->Reset();

cpp/src/arrow/util/uri.h

@@ -54,8 +54,9 @@ class ARROW_EXPORT Uri {
   int32_t port() const;
   /// The URI path component.
   std::string path() const;
+
   /// Get the string representation of this URI.
-  const std::string& to_string() const;
+  const std::string& ToString() const;
 
   /// Factory function to parse a URI from its string representation.
   Status Parse(const std::string& uri_string);

python/examples/flight/client.py

@@ -90,6 +90,8 @@ def get_flight(args, client):
 
 
 def _add_common_arguments(parser):
+    parser.add_argument('--tls', action='store_true')
+    parser.add_argument('--tls-roots', default=None)
     parser.add_argument('host', type=str,
                         help="The host to connect to.")
 
@@ -131,7 +133,15 @@ def main():
     }
     host, port = args.host.split(':')
     port = int(port)
-    client = pyarrow.flight.FlightClient.connect(f"grpc://{host}:{port}")
+    scheme = "grpc+tcp"
+    connection_args = {}
+    if args.tls:
+        scheme = "grpc+tls"
+        if args.tls_roots:
+            with open(args.tls_roots, "rb") as root_certs:
+                connection_args["tls_root_certs"] = root_certs.read()
+    client = pyarrow.flight.FlightClient.connect(f"{scheme}://{host}:{port}",
+                                                 **connection_args)
     while True:
         try:
             action = pyarrow.flight.Action("healthcheck", b"")

python/examples/flight/server.py

@@ -17,6 +17,7 @@
 
 """An example Flight Python server."""
 
+import argparse
 import ast
 import threading
 import time
@@ -76,7 +77,7 @@ class FlightServer(pyarrow.flight.FlightServerBase):
             return None
         return pyarrow.flight.RecordBatchStream(self.flights[key])
 
-    def list_actions(self, context):
+    def list_actions(self):
         return [
             ("clear", "Clear the stored flights."),
             ("shutdown", "Shut down this server."),
@@ -100,8 +101,25 @@ class FlightServer(pyarrow.flight.FlightServerBase):
 
 
 def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, default=5005)
+    parser.add_argument("--tls", nargs=2, default=None)
+
+    args = parser.parse_args()
+
     server = FlightServer()
-    server.run(5005)
+    kwargs = {}
+    scheme = "grpc+tcp"
+    if args.tls:
+        scheme = "grpc+tls"
+        with open(args.tls[0], "rb") as cert_file:
+            kwargs["tls_cert_chain"] = cert_file.read()
+        with open(args.tls[1], "rb") as key_file:
+            kwargs["tls_private_key"] = key_file.read()
+
+    location = "{}://0.0.0.0:{}".format(scheme, args.port)
+    print("Serving on", location)
+    server.run(location, **kwargs)
 
 
 if __name__ == '__main__':

python/pyarrow/_flight.pyx

@@ -72,7 +72,7 @@ cdef class Action:
         return pyarrow_wrap_buffer(self.action.body)
 
     @staticmethod
-    cdef CAction unwrap(action):
+    cdef CAction unwrap(action) except *:
         if not isinstance(action, Action):
             raise TypeError("Must provide Action, not '{}'".format(
                 type(action)))
@@ -168,11 +168,11 @@ cdef class FlightDescriptor:
         return "<FlightDescriptor type: {!r}>".format(self.descriptor_type)
 
     @staticmethod
-    cdef FlightDescriptor unwrap(descriptor):
+    cdef CFlightDescriptor unwrap(descriptor) except *:
         if not isinstance(descriptor, FlightDescriptor):
             raise TypeError("Must provide a FlightDescriptor, not '{}'".format(
                 type(descriptor)))
-        return <FlightDescriptor> descriptor
+        return (<FlightDescriptor> descriptor).descriptor
 
 
 cdef class Ticket:
@@ -225,9 +225,9 @@ cdef class Location:
         return result
 
     @staticmethod
-    cdef CLocation unwrap(object location):
+    cdef CLocation unwrap(object location) except *:
         cdef CLocation c_location
-        if isinstance(location, (six.text_type, six.binary_type)):
+        if isinstance(location, six.text_type):
             check_status(CLocation.Parse(tobytes(location), &c_location))
             return c_location
         elif not isinstance(location, Location):
@@ -381,21 +381,20 @@ cdef class FlightClient:
                         .format(self.__class__.__name__))
 
     @staticmethod
-    def connect(location):
+    def connect(location, **kwargs):
         """Connect to a Flight service on the given host and port."""
         cdef:
             FlightClient result = FlightClient.__new__(FlightClient)
             int c_port = 0
-            CLocation c_location
+            CLocation c_location = Location.unwrap(location)
+            CFlightClientOptions c_options
 
-        if isinstance(location, Location):
-            c_location = (<Location> location).location
-        else:
-            c_location = CLocation()
-            check_status(CLocation.Parse(tobytes(location), &c_location))
+        if "tls_root_certs" in kwargs:
+            c_options.tls_root_certs = tobytes(kwargs["tls_root_certs"])
 
         with nogil:
-            check_status(CFlightClient.Connect(c_location, &result.client))
+            check_status(CFlightClient.Connect(c_location, c_options,
+                                               &result.client))
 
         return result
 
@@ -478,12 +477,12 @@ cdef class FlightClient:
         cdef:
             FlightInfo result = FlightInfo.__new__(FlightInfo)
             CFlightCallOptions* c_options = FlightCallOptions.unwrap(options)
-            FlightDescriptor c_descriptor = \
+            CFlightDescriptor c_descriptor = \
                 FlightDescriptor.unwrap(descriptor)
 
         with nogil:
             check_status(self.client.get().GetFlightInfo(
-                deref(c_options), c_descriptor.descriptor, &result.info))
+                deref(c_options), c_descriptor, &result.info))
 
         return result
 
@@ -494,7 +493,8 @@ cdef class FlightClient:
             CFlightCallOptions* c_options = FlightCallOptions.unwrap(options)
 
         with nogil:
-            check_status(self.client.get().DoGet(deref(c_options), ticket.ticket, &reader))
+            check_status(self.client.get().DoGet(
+                deref(c_options), ticket.ticket, &reader))
         result = FlightRecordBatchReader()
         result.reader.reset(reader.release())
         return result
@@ -506,12 +506,12 @@ cdef class FlightClient:
             shared_ptr[CSchema] c_schema = pyarrow_unwrap_schema(schema)
             unique_ptr[CRecordBatchWriter] writer
             CFlightCallOptions* c_options = FlightCallOptions.unwrap(options)
-            FlightDescriptor c_descriptor = \
+            CFlightDescriptor c_descriptor = \
                 FlightDescriptor.unwrap(descriptor)
 
         with nogil:
             check_status(self.client.get().DoPut(
-                deref(c_options), c_descriptor.descriptor, c_schema, &writer))
+                deref(c_options), c_descriptor, c_schema, &writer))
         result = FlightRecordBatchWriter()
         result.writer.reset(writer.release())
         return result
@@ -520,7 +520,7 @@ cdef class FlightClient:
 cdef class FlightDataStream:
     """Abstract base class for Flight data streams."""
 
-    cdef CFlightDataStream* to_stream(self):
+    cdef CFlightDataStream* to_stream(self) except *:
         """Create the C++ data stream for the backing Python object.
 
         We don't expose the C++ object to Python, so we can manage its
@@ -547,7 +547,7 @@ cdef class RecordBatchStream(FlightDataStream):
                             "but got: {}".format(type(data_source)))
         self.data_source = data_source
 
-    cdef CFlightDataStream* to_stream(self):
+    cdef CFlightDataStream* to_stream(self) except *:
         cdef:
             shared_ptr[CRecordBatchReader] reader
         if isinstance(self.data_source, _CRecordBatchReader):
@@ -585,7 +585,7 @@ cdef class GeneratorStream(FlightDataStream):
         self.schema = pyarrow_unwrap_schema(schema)
         self.generator = iter(generator)
 
-    cdef CFlightDataStream* to_stream(self):
+    cdef CFlightDataStream* to_stream(self) except *:
         cdef:
             function[cb_data_stream_next] callback = &_data_stream_next
         return new CPyGeneratorFlightDataStream(self, self.schema, callback)
@@ -985,20 +985,30 @@ cdef class FlightServerBase:
     cdef:
         unique_ptr[PyFlightServer] server
 
-    def run(self, location, auth_handler=None):
+    def run(self, location, auth_handler=None, **kwargs):
         cdef:
             PyFlightServerVtable vtable = PyFlightServerVtable()
             PyFlightServer* c_server
-            CLocation c_location = Location.unwrap(location)
-            unique_ptr[CServerAuthHandler] c_auth_handler
+            unique_ptr[CFlightServerOptions] c_options
+
+        c_options.reset(new CFlightServerOptions(Location.unwrap(location)))
 
         if auth_handler:
             if not isinstance(auth_handler, ServerAuthHandler):
                 raise TypeError("auth_handler must be a ServerAuthHandler, "
                                 "not a '{}'".format(type(auth_handler)))
-            c_auth_handler.reset(
+            c_options.get().auth_handler.reset(
                 (<ServerAuthHandler> auth_handler).to_handler())
 
+        if "tls_cert_chain" in kwargs:
+            if "tls_private_key" not in kwargs:
+                raise ValueError(
+                    "Must provide both cert chain and private key")
+            c_options.get().tls_cert_chain = tobytes(
+                kwargs["tls_cert_chain"])
+            c_options.get().tls_private_key = tobytes(
+                kwargs["tls_private_key"])
+
         vtable.list_flights = &_list_flights
         vtable.get_flight_info = &_get_flight_info
         vtable.do_put = &_do_put
@@ -1009,7 +1019,7 @@ cdef class FlightServerBase:
         c_server = new PyFlightServer(self, vtable)
         self.server.reset(c_server)
         with nogil:
-            check_status(c_server.Init(move(c_auth_handler), c_location))
+            check_status(c_server.Init(deref(c_options)))
             check_status(c_server.ServeWithSignals())
 
     def list_flights(self, context, criteria):

python/pyarrow/includes/libarrow_flight.pxd

@@ -153,9 +153,21 @@ cdef extern from "arrow/flight/api.h" namespace "arrow" nogil:
         CFlightCallOptions()
         CTimeoutDuration timeout
 
+    cdef cppclass CFlightServerOptions" arrow::flight::FlightServerOptions":
+        CFlightServerOptions(const CLocation& location)
+        CLocation location
+        unique_ptr[CServerAuthHandler] auth_handler
+        c_string tls_cert_chain
+        c_string tls_private_key
+
+    cdef cppclass CFlightClientOptions" arrow::flight::FlightClientOptions":
+        CFlightClientOptions()
+        c_string tls_root_certs
+
     cdef cppclass CFlightClient" arrow::flight::FlightClient":
         @staticmethod
         CStatus Connect(const CLocation& location,
+                        const CFlightClientOptions& options,
                         unique_ptr[CFlightClient]* client)
 
         CStatus Authenticate(CFlightCallOptions& options,
@@ -229,8 +241,7 @@ cdef extern from "arrow/python/flight.h" namespace "arrow::py::flight" nogil:
     cdef cppclass PyFlightServer:
         PyFlightServer(object server, PyFlightServerVtable vtable)
 
-        CStatus Init(unique_ptr[CServerAuthHandler] auth_handler,
-                     const CLocation& location)
+        CStatus Init(CFlightServerOptions& options)
         CStatus ServeWithSignals() except *
         void Shutdown()
 

python/pyarrow/tests/test_flight.py

@@ -397,3 +397,13 @@ def test_token_auth_invalid():
         client = flight.FlightClient.connect(server_location)
         with pytest.raises(pa.ArrowException, match=".*unauthenticated.*"):
             client.authenticate(TokenClientAuthHandler('test', 'wrong'))
+
+
+def test_location_invalid():
+    """Test constructing invalid URIs."""
+    with pytest.raises(pa.ArrowException, match=".*Cannot parse URI:.*"):
+        flight.FlightClient.connect("%")
+
+    server = ConstantFlightServer()
+    with pytest.raises(pa.ArrowException, match=".*Cannot parse URI:.*"):
+        server.run("%")