Taking a stand on request nullability in river v2

Author: blast-hardcheese

src/replit_river/codegen/client.py

@@ -978,32 +978,9 @@ async def {name}(
                 ]
             )
         elif protocol_version == "v2.0":
-            assert init_meta, "Protocol v2 requires init to be defined"
-            _, init_type, render_init_method = init_meta
-            current_chunks.extend(
-                [
-                    reindent(
-                        "  ",
-                        f"""\
-        async def {name}(
-          self,
-          init: {render_type_expr(init_type)},
-        ) -> {  # TODO(dstewart) This should just be output_type
-                            render_type_expr(output_or_error_type)
-                        }:
-          return await self.client.send_upload(
-            {repr(schema_name)},
-            {repr(name)},
-            init,
-            None,
-            {reindent("                    ", render_init_method)},
-            None,
-            {reindent("                    ", parse_output_method)},
-            {reindent("                    ", parse_error_method)},
-          )
-                    """,
-                    )
-                ]
+            raise ValueError(
+                "It is expected that protocol v2 uploads have both init and input "
+                "defined, otherwise it's no different than rpc",
             )
         else:
             assert_never(protocol_version)

src/replit_river/v2/client.py

@@ -75,9 +75,9 @@ async def send_upload[I, R, A](
         service_name: str,
         procedure_name: str,
         init: I,
-        request: AsyncIterable[R] | None,
+        request: AsyncIterable[R],
         init_serializer: Callable[[I], Any],
-        request_serializer: Callable[[R], Any] | None,
+        request_serializer: Callable[[R], Any],
         response_deserializer: Callable[[Any], A],
         error_deserializer: Callable[[Any], RiverError],
     ) -> A:

src/replit_river/v2/session.py

@@ -661,9 +661,9 @@ async def send_upload[I, R, A](
         service_name: str,
         procedure_name: str,
         init: I,
-        request: AsyncIterable[R] | None,
+        request: AsyncIterable[R],
         init_serializer: Callable[[I], Any],
-        request_serializer: Callable[[R], Any] | None,
+        request_serializer: Callable[[R], Any],
         response_deserializer: Callable[[Any], A],
         error_deserializer: Callable[[Any], RiverError],
         span: Span,
@@ -684,25 +684,22 @@ async def send_upload[I, R, A](
 
         async with self._with_stream(stream_id, 1) as (backpressure_waiter, output):
             try:
-                if request:
-                    assert request_serializer, "send_stream missing request_serializer"
-
-                    # If this request is not closed and the session is killed, we should
-                    # throw exception here
-                    async for item in request:
-                        # Block for backpressure
-                        await backpressure_waiter.wait()
-                        if output.closed():
-                            logger.debug("Stream is closed, avoid sending the rest")
-                            break
-                        await self._send_message(
-                            stream_id=stream_id,
-                            service_name=service_name,
-                            procedure_name=procedure_name,
-                            control_flags=0,
-                            payload=request_serializer(item),
-                            span=span,
-                        )
+                # If this request is not closed and the session is killed, we should
+                # throw exception here
+                async for item in request:
+                    # Block for backpressure
+                    await backpressure_waiter.wait()
+                    if output.closed():
+                        logger.debug("Stream is closed, avoid sending the rest")
+                        break
+                    await self._send_message(
+                        stream_id=stream_id,
+                        service_name=service_name,
+                        procedure_name=procedure_name,
+                        control_flags=0,
+                        payload=request_serializer(item),
+                        span=span,
+                    )
             except WebsocketClosedException as e:
                 raise RiverServiceException(
                     ERROR_CODE_STREAM_CLOSED, str(e), service_name, procedure_name