Class GetInfoFlightServer

python/pyarrow/tests/test_flight.py:285–309

A Flight server that tests GetFlightInfo.

class GetInfoFlightServer(FlightServerBase):
    """A Flight server that tests GetFlightInfo."""

    def get_flight_info(self, context, descriptor):
        return flight.FlightInfo(
            pa.schema([('a', pa.int32())]),
            descriptor,
            [
                flight.FlightEndpoint(b'', ['grpc://test']),
                flight.FlightEndpoint(
                    b'',
                    [flight.Location.for_grpc_tcp('localhost', 5005)],
                    pa.scalar("2023-04-05T12:34:56.789012345").cast(pa.timestamp("ns")),
                    "endpoint app metadata"
                ),
            ],
            1,
            42,
            True,
            "info app metadata"
        )

    def get_schema(self, context, descriptor):
        info = self.get_flight_info(context, descriptor)
        return flight.SchemaResult(info.schema)


class ListActionsFlightServer(FlightServerBase):