swadhin / justapi public

swadhinbiswas

[email protected]
2026-07-09 11:30:00 +0600 committed NaN years ago
43e015f
commit 43e015f
23 files changed
+595 -0
Showing 23 changed files+595-0
tests/__init__.pytests/proto/__init__.pytests/proto/gen/__init__.pytests/proto/gen/helloworld_pb2.pytests/proto/gen/helloworld_pb2.pyitests/proto/gen/helloworld_pb2_grpc.pytests/proto/helloworld.prototests/test_annotations.pytests/test_asyncio.pytests/test_batching.pytests/test_buffer.pytests/test_deadlock.pytests/test_dump_threads.pytests/test_exceptions.pytests/test_grpc_client.pytests/test_grpc_inspect.pytests/test_grpc_interceptor.pytests/test_grpc_server.pytests/test_loop_alive.pytests/test_middleware.pytests/test_multipart.pytests/test_responses.pytests/test_sse.py
▼tests/__init__.py
+1-0
@@ -0,0 +1 @@
1
+# init file
0
2
▼tests/proto/__init__.py
+1-0
@@ -0,0 +1 @@
1
+# init file
0
2
▼tests/proto/gen/__init__.py
+1-0
@@ -0,0 +1 @@
1
+# init file
0
2
▼tests/proto/gen/helloworld_pb2.py
+40-0
@@ -0,0 +1,40 @@
1
+# -*- coding: utf-8 -*-
2
+# Generated by the protocol buffer compiler. DO NOT EDIT!
3
+# NO CHECKED-IN PROTOBUF GENCODE
4
+# source: helloworld.proto
5
+# Protobuf Python Version: 6.33.5
6
+"""Generated protocol buffer code."""
7
+from google.protobuf import descriptor as _descriptor
8
+from google.protobuf import descriptor_pool as _descriptor_pool
9
+from google.protobuf import runtime_version as _runtime_version
10
+from google.protobuf import symbol_database as _symbol_database
11
+from google.protobuf.internal import builder as _builder
12
+_runtime_version.ValidateProtobufRuntimeVersion(
13
+ _runtime_version.Domain.PUBLIC,
14
+ 6,
15
+ 33,
16
+ 5,
17
+ '',
18
+ 'helloworld.proto'
19
+)
20
+# @@protoc_insertion_point(imports)
21
+
22
+_sym_db = _symbol_database.Default()
23
+
24
+
25
+
26
+
27
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10helloworld.proto\x12\nhelloworld\"\x1c\n\x0cHelloRequest\x12\x0c\n\x04name\x18\x01 \x01(\t\"\x1d\n\nHelloReply\x12\x0f\n\x07message\x18\x01 \x01(\t2I\n\x07Greeter\x12>\n\x08SayHello\x12\x18.helloworld.HelloRequest\x1a\x16.helloworld.HelloReply\"\x00\x62\x06proto3')
28
+
29
+_globals = globals()
30
+_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
31
+_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'helloworld_pb2', _globals)
32
+if not _descriptor._USE_C_DESCRIPTORS:
33
+ DESCRIPTOR._loaded_options = None
34
+ _globals['_HELLOREQUEST']._serialized_start=32
35
+ _globals['_HELLOREQUEST']._serialized_end=60
36
+ _globals['_HELLOREPLY']._serialized_start=62
37
+ _globals['_HELLOREPLY']._serialized_end=91
38
+ _globals['_GREETER']._serialized_start=93
39
+ _globals['_GREETER']._serialized_end=166
40
+# @@protoc_insertion_point(module_scope)
0
41
▼tests/proto/gen/helloworld_pb2.pyi
+17-0
@@ -0,0 +1,17 @@
1
+from google.protobuf import descriptor as _descriptor
2
+from google.protobuf import message as _message
3
+from typing import ClassVar as _ClassVar, Optional as _Optional
4
+
5
+DESCRIPTOR: _descriptor.FileDescriptor
6
+
7
+class HelloRequest(_message.Message):
8
+ __slots__ = ("name",)
9
+ NAME_FIELD_NUMBER: _ClassVar[int]
10
+ name: str
11
+ def __init__(self, name: _Optional[str] = ...) -> None: ...
12
+
13
+class HelloReply(_message.Message):
14
+ __slots__ = ("message",)
15
+ MESSAGE_FIELD_NUMBER: _ClassVar[int]
16
+ message: str
17
+ def __init__(self, message: _Optional[str] = ...) -> None: ...
0
18
▼tests/proto/gen/helloworld_pb2_grpc.py
+97-0
@@ -0,0 +1,97 @@
1
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
2
+"""Client and server classes corresponding to protobuf-defined services."""
3
+import grpc
4
+import warnings
5
+
6
+from . import helloworld_pb2 as helloworld__pb2
7
+
8
+GRPC_GENERATED_VERSION = '1.81.1'
9
+GRPC_VERSION = grpc.__version__
10
+_version_not_supported = False
11
+
12
+try:
13
+ from grpc._utilities import first_version_is_lower
14
+ _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
15
+except ImportError:
16
+ _version_not_supported = True
17
+
18
+if _version_not_supported:
19
+ raise RuntimeError(
20
+ f'The grpc package installed is at version {GRPC_VERSION},'
21
+ + ' but the generated code in helloworld_pb2_grpc.py depends on'
22
+ + f' grpcio>={GRPC_GENERATED_VERSION}.'
23
+ + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
24
+ + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
25
+ )
26
+
27
+
28
+class GreeterStub:
29
+ """Missing associated documentation comment in .proto file."""
30
+
31
+ def __init__(self, channel):
32
+ """Constructor.
33
+
34
+ Args:
35
+ channel: A grpc.Channel.
36
+ """
37
+ self.SayHello = channel.unary_unary(
38
+ '/helloworld.Greeter/SayHello',
39
+ request_serializer=helloworld__pb2.HelloRequest.SerializeToString,
40
+ response_deserializer=helloworld__pb2.HelloReply.FromString,
41
+ _registered_method=True)
42
+
43
+
44
+class GreeterServicer:
45
+ """Missing associated documentation comment in .proto file."""
46
+
47
+ def SayHello(self, request, context):
48
+ """Missing associated documentation comment in .proto file."""
49
+ context.set_code(grpc.StatusCode.UNIMPLEMENTED)
50
+ context.set_details('Method not implemented!')
51
+ raise NotImplementedError('Method not implemented!')
52
+
53
+
54
+def add_GreeterServicer_to_server(servicer, server):
55
+ rpc_method_handlers = {
56
+ 'SayHello': grpc.unary_unary_rpc_method_handler(
57
+ servicer.SayHello,
58
+ request_deserializer=helloworld__pb2.HelloRequest.FromString,
59
+ response_serializer=helloworld__pb2.HelloReply.SerializeToString,
60
+ ),
61
+ }
62
+ generic_handler = grpc.method_handlers_generic_handler(
63
+ 'helloworld.Greeter', rpc_method_handlers)
64
+ server.add_generic_rpc_handlers((generic_handler,))
65
+ server.add_registered_method_handlers('helloworld.Greeter', rpc_method_handlers)
66
+
67
+
68
+ # This class is part of an EXPERIMENTAL API.
69
+class Greeter:
70
+ """Missing associated documentation comment in .proto file."""
71
+
72
+ @staticmethod
73
+ def SayHello(request,
74
+ target,
75
+ options=(),
76
+ channel_credentials=None,
77
+ call_credentials=None,
78
+ insecure=False,
79
+ compression=None,
80
+ wait_for_ready=None,
81
+ timeout=None,
82
+ metadata=None):
83
+ return grpc.experimental.unary_unary(
84
+ request,
85
+ target,
86
+ '/helloworld.Greeter/SayHello',
87
+ helloworld__pb2.HelloRequest.SerializeToString,
88
+ helloworld__pb2.HelloReply.FromString,
89
+ options,
90
+ channel_credentials,
91
+ insecure,
92
+ call_credentials,
93
+ compression,
94
+ wait_for_ready,
95
+ timeout,
96
+ metadata,
97
+ _registered_method=True)
0
98
▼tests/proto/helloworld.proto
+15-0
@@ -0,0 +1,15 @@
1
+syntax = "proto3";
2
+
3
+package helloworld;
4
+
5
+service Greeter {
6
+ rpc SayHello (HelloRequest) returns (HelloReply) {}
7
+}
8
+
9
+message HelloRequest {
10
+ string name = 1;
11
+}
12
+
13
+message HelloReply {
14
+ string message = 1;
15
+}
0
16
▼tests/test_annotations.py
+26-0
@@ -0,0 +1,26 @@
1
+import threading
2
+import time
3
+import urllib.request
4
+from justapi import JustAPIApp, Path, Query, Header, JSONResponse
5
+
6
+app = JustAPIApp()
7
+
8
+@app.get("/hello/{name}")
9
+def hello(name: str = Path(), count: int = Query(1), x_token: str = Header(alias="X-Token", default="abc")):
10
+ return JSONResponse({
11
+ "name": name,
12
+ "count": count,
13
+ "token": x_token,
14
+ })
15
+
16
+def run_server():
17
+ app.run("127.0.0.1:8083")
18
+
19
+if __name__ == "__main__":
20
+ t = threading.Thread(target=run_server, daemon=True)
21
+ t.start()
22
+ time.sleep(1)
23
+
24
+ req = urllib.request.Request("http://127.0.0.1:8083/hello/world?count=5", headers={"X-Token": "xyz"})
25
+ with urllib.request.urlopen(req) as response:
26
+ print(response.read().decode())
0
27
▼tests/test_asyncio.py
+17-0
@@ -0,0 +1,17 @@
1
+import asyncio
2
+import threading
3
+
4
+_loop = asyncio.new_event_loop()
5
+
6
+def start_loop(loop):
7
+ asyncio.set_event_loop(loop)
8
+ loop.run_forever()
9
+
10
+t = threading.Thread(target=start_loop, args=(_loop,), daemon=True)
11
+t.start()
12
+
13
+async def my_coro():
14
+ return 42
15
+
16
+future = asyncio.run_coroutine_threadsafe(my_coro(), _loop)
17
+print(future.result())
0
18
▼tests/test_batching.py
+52-0
@@ -0,0 +1,52 @@
1
+import asyncio
2
+import time
3
+from justapi.app import JustAPIApp, adaptive_batch
4
+import urllib.request
5
+import threading
6
+
7
+app = JustAPIApp()
8
+
9
+@app.post("/predict")
10
+@adaptive_batch(max_size=3, window_ms=50)
11
+def predict(requests):
12
+ # Log the size of the batch
13
+ print(f"Received batch of size {len(requests)}")
14
+ # Simulate ML model computation
15
+ time.sleep(0.1)
16
+
17
+ results = []
18
+ for req in requests:
19
+ body = req.get("body", b"").decode("utf-8")
20
+ results.append({"status": 200, "body": f"Processed {body}"})
21
+ return results
22
+
23
+def test_batching():
24
+ def run_server():
25
+ try:
26
+ app.run("127.0.0.1:8081")
27
+ except Exception:
28
+ pass
29
+
30
+ server_thread = threading.Thread(target=run_server, daemon=True)
31
+ server_thread.start()
32
+
33
+ time.sleep(1)
34
+
35
+ def make_request(idx):
36
+ req = urllib.request.Request("http://127.0.0.1:8081/predict", data=f"req-{idx}".encode("utf-8"), method="POST")
37
+ try:
38
+ with urllib.request.urlopen(req) as response:
39
+ return response.read().decode("utf-8")
40
+ except Exception as e:
41
+ return str(e)
42
+
43
+ threads = []
44
+ for i in range(5):
45
+ t = threading.Thread(target=lambda i=i: print(f"Response {i}: {make_request(i)}"))
46
+ threads.append(t)
47
+ t.start()
48
+
49
+ for t in threads:
50
+ t.join()
51
+
52
+ print("Done")
0
53
▼tests/test_buffer.py
+11-0
@@ -0,0 +1,11 @@
1
+import sys
2
+sys.path.insert(0, "/home/swadhin/RastAPI/crates/justapi-py")
3
+import justapi
4
+
5
+print("Testing buffer...")
6
+try:
7
+ buf = justapi._justapi._test_zero_copy(b"Hello from Rust Zero-Copy Buffer!")
8
+ mv = memoryview(buf)
9
+ print("Memoryview created successfully:", mv.tobytes())
10
+except Exception as e:
11
+ print("Failed:", e)
0
12
▼tests/test_deadlock.py
+9-0
@@ -0,0 +1,9 @@
1
+from justapi import JustAPIApp
2
+import asyncio
3
+app = JustAPIApp()
4
+async def hello(req):
5
+ await asyncio.sleep(0.01)
6
+ return {"message": "hello"}
7
+app.get("/", hello)
8
+if __name__ == "__main__":
9
+ app.run("127.0.0.1:8080")
0
10
▼tests/test_dump_threads.py
+25-0
@@ -0,0 +1,25 @@
1
+from justapi import JustAPIApp
2
+import asyncio, threading, sys, traceback
3
+
4
+app = JustAPIApp()
5
+
6
+async def hello(req):
7
+ await asyncio.sleep(0.01)
8
+ return {"message": "hello"}
9
+
10
+app.get("/", hello)
11
+
12
+def dump_threads():
13
+ import time
14
+ while True:
15
+ time.sleep(2)
16
+ print("--- THREAD DUMP ---")
17
+ for th in threading.enumerate():
18
+ print(th)
19
+ traceback.print_stack(sys._current_frames()[th.ident])
20
+ print("-------------------")
21
+
22
+threading.Thread(target=dump_threads, daemon=True).start()
23
+
24
+if __name__ == "__main__":
25
+ app.run("127.0.0.1:8080")
0
26
▼tests/test_exceptions.py
+31-0
@@ -0,0 +1,31 @@
1
+import threading
2
+import time
3
+import urllib.request
4
+import urllib.error
5
+from justapi import JustAPIApp, Path, RequestValidationError, JSONResponse
6
+
7
+app = JustAPIApp()
8
+
9
+@app.get("/items/{item_id}")
10
+def get_item(item_id: int = Path()):
11
+ return {"item": item_id}
12
+
13
+def validation_exception_handler(request, exc):
14
+ return JSONResponse({"detail": exc.errors}, status_code=422)
15
+
16
+app.add_exception_handler(RequestValidationError, validation_exception_handler)
17
+
18
+def run_server():
19
+ app.run("127.0.0.1:8084")
20
+
21
+if __name__ == "__main__":
22
+ t = threading.Thread(target=run_server, daemon=True)
23
+ t.start()
24
+ time.sleep(1)
25
+
26
+ try:
27
+ req = urllib.request.Request("http://127.0.0.1:8084/items/abc")
28
+ with urllib.request.urlopen(req) as response:
29
+ print("Success:", response.read().decode())
30
+ except urllib.error.HTTPError as e:
31
+ print(f"Error {e.code}: {e.read().decode()}")
0
32
▼tests/test_grpc_client.py
+12-0
@@ -0,0 +1,12 @@
1
+import grpc
2
+from tests.proto.gen import helloworld_pb2, helloworld_pb2_grpc
3
+
4
+def run():
5
+ with grpc.insecure_channel('127.0.0.1:50051') as channel:
6
+ stub = helloworld_pb2_grpc.GreeterStub(channel)
7
+ print("Calling SayHello...")
8
+ response = stub.SayHello(helloworld_pb2.HelloRequest(name='JustAPI'))
9
+ print("Greeter client received: " + response.message)
10
+
11
+if __name__ == '__main__':
12
+ run()
0
13
▼tests/test_grpc_inspect.py
+12-0
@@ -0,0 +1,12 @@
1
+from justapi.app import JustAPIApp
2
+from tests.proto.gen import helloworld_pb2, helloworld_pb2_grpc
3
+
4
+class GreeterServicer(helloworld_pb2_grpc.GreeterServicer):
5
+ def SayHello(self, request, context):
6
+ return helloworld_pb2.HelloReply(message="Hello from Python!")
7
+
8
+def test_grpc_inspect():
9
+ app = JustAPIApp()
10
+ app.set_grpc_addr("127.0.0.1:50051")
11
+ app.add_grpc_service(GreeterServicer(), helloworld_pb2_grpc.add_GreeterServicer_to_server)
12
+ print("gRPC service registered successfully!")
0
13
▼tests/test_grpc_interceptor.py
+1-0
@@ -0,0 +1 @@
1
+# Just a placeholder script
0
2
▼tests/test_grpc_server.py
+14-0
@@ -0,0 +1,14 @@
1
+from justapi.app import JustAPIApp
2
+from tests.proto.gen import helloworld_pb2, helloworld_pb2_grpc
3
+
4
+class GreeterServicer(helloworld_pb2_grpc.GreeterServicer):
5
+ def SayHello(self, request, context):
6
+ print(f"Received request in Python: {request.name}")
7
+ return helloworld_pb2.HelloReply(message=f"Hello from Python, {request.name}!")
8
+
9
+def test_grpc_server_setup():
10
+ app = JustAPIApp()
11
+ app.set_grpc_addr("127.0.0.1:50051")
12
+ app.add_grpc_service(GreeterServicer(), helloworld_pb2_grpc.add_GreeterServicer_to_server)
13
+ print("Starting JustAPI with gRPC on :50051...")
14
+ # app.run("127.0.0.1:8000") # Do not run in pytest
0
15
▼tests/test_loop_alive.py
+21-0
@@ -0,0 +1,21 @@
1
+from justapi import JustAPIApp
2
+import asyncio, threading
3
+
4
+app = JustAPIApp()
5
+
6
+async def hello(req):
7
+ await asyncio.sleep(0.01)
8
+ return {"message": "hello"}
9
+
10
+app.get("/", hello)
11
+
12
+def monitor():
13
+ import time
14
+ while True:
15
+ time.sleep(1)
16
+ print("Threads:", threading.enumerate())
17
+
18
+threading.Thread(target=monitor, daemon=True).start()
19
+
20
+if __name__ == "__main__":
21
+ app.run("127.0.0.1:8080")
0
22
▼tests/test_middleware.py
+94-0
@@ -0,0 +1,94 @@
1
+import asyncio
2
+import httpx
3
+from justapi import JustAPIApp, APIRouter
4
+
5
+app = JustAPIApp()
6
+router = APIRouter(prefix="/router")
7
+
8
+@app.middleware("http")
9
+async def add_process_time_header(request, call_next):
10
+ assert request.method == "GET"
11
+ assert request.path.startswith("/test") or request.path.startswith("/router")
12
+
13
+ response = await call_next(request)
14
+
15
+ if "headers" not in response:
16
+ response["headers"] = []
17
+
18
+ response["headers"].append((b"X-Process-Time", b"0.123"))
19
+ return response
20
+
21
+@app.get("/test_async")
22
+async def get_test_async():
23
+ return {"message": "async success"}
24
+
25
+async def route_middleware(request, call_next):
26
+ res = await call_next(request)
27
+ res["headers"].append((b"X-Route-Mw", b"route"))
28
+ return res
29
+
30
+@app.get("/test_route_mw", middlewares=[route_middleware])
31
+def get_test_route_mw():
32
+ return {"message": "route mw"}
33
+
34
+async def router_middleware(request, call_next):
35
+ res = await call_next(request)
36
+ res["headers"].append((b"X-Router-Mw", b"router"))
37
+ return res
38
+
39
+router.middlewares.append(router_middleware)
40
+
41
+@router.get("/test", middlewares=[route_middleware])
42
+def get_router_test():
43
+ return {"message": "router mw"}
44
+
45
+app.include_router(router)
46
+
47
+@app.get("/test_sync")
48
+def get_test_sync():
49
+ return {"message": "sync success"}
50
+
51
+async def main():
52
+ import threading
53
+ import time
54
+
55
+ def run_server():
56
+ app.run("127.0.0.1:8081")
57
+
58
+ t = threading.Thread(target=run_server, daemon=True)
59
+ t.start()
60
+ time.sleep(1) # wait for server to start
61
+
62
+ async with httpx.AsyncClient() as client:
63
+ # Test Async
64
+ res1 = await client.get('http://localhost:8081/test_async')
65
+ print(res1.status_code, res1.headers, res1.json())
66
+ assert res1.status_code == 200
67
+ assert res1.json() == {"message": "async success"}
68
+ assert res1.headers.get("x-process-time") == "0.123"
69
+
70
+ # Test Sync
71
+ res2 = await client.get('http://localhost:8081/test_sync')
72
+ print(res2.status_code, res2.headers, res2.json())
73
+ assert res2.status_code == 200
74
+ assert res2.json() == {"message": "sync success"}
75
+ assert res2.headers.get("x-process-time") == "0.123"
76
+
77
+ # Test Route MW
78
+ res3 = await client.get('http://localhost:8081/test_route_mw')
79
+ assert res3.status_code == 200
80
+ assert res3.headers.get("x-route-mw") == "route"
81
+ assert res3.headers.get("x-process-time") == "0.123"
82
+
83
+ # Test Router MW
84
+ res4 = await client.get('http://localhost:8081/router/test')
85
+ print("RES4:", res4.status_code, res4.headers, res4.content)
86
+ assert res4.status_code == 200
87
+ assert res4.headers.get("x-route-mw") == "route"
88
+ assert res4.headers.get("x-router-mw") == "router"
89
+ assert res4.headers.get("x-process-time") == "0.123"
90
+
91
+ print("Middleware test passed!")
92
+
93
+if __name__ == "__main__":
94
+ asyncio.run(main())
0
95
▼tests/test_multipart.py
+37-0
@@ -0,0 +1,37 @@
1
+import asyncio
2
+import httpx
3
+from justapi import JustAPIApp, UploadFile, File, Form
4
+
5
+app = JustAPIApp()
6
+
7
+@app.post("/upload")
8
+def upload_file(file: UploadFile = File(), username: str = Form()):
9
+ content = file.read()
10
+ file.close()
11
+ return {"filename": file.filename, "content": content.decode("utf-8"), "username": username}
12
+
13
+async def main():
14
+ # Start server in background
15
+ import threading
16
+ import time
17
+
18
+ def run_server():
19
+ app.run("127.0.0.1:8080")
20
+
21
+ t = threading.Thread(target=run_server, daemon=True)
22
+ t.start()
23
+ time.sleep(1) # wait for server to start
24
+
25
+ async with httpx.AsyncClient() as client:
26
+ files = {'file': ('test.txt', b'Hello, world!', 'text/plain')}
27
+ data = {'username': 'testuser'}
28
+ response = await client.post('http://localhost:8080/upload', files=files, data=data)
29
+
30
+ print(response.status_code)
31
+ print(response.json())
32
+ assert response.status_code == 200
33
+ assert response.json() == {"filename": "test.txt", "content": "Hello, world!", "username": "testuser"}
34
+ print("Multipart upload test passed!")
35
+
36
+if __name__ == "__main__":
37
+ asyncio.run(main())
0
38
▼tests/test_responses.py
+49-0
@@ -0,0 +1,49 @@
1
+import threading
2
+import time
3
+import urllib.request
4
+from justapi import JustAPIApp, JSONResponse, HTMLResponse, RedirectResponse, PlainTextResponse
5
+
6
+app = JustAPIApp()
7
+
8
+@app.get("/json")
9
+def get_json():
10
+ return JSONResponse({"hello": "world"}, status_code=201)
11
+
12
+@app.get("/html")
13
+def get_html():
14
+ return HTMLResponse("<h1>Hello</h1>")
15
+
16
+@app.get("/redirect")
17
+def get_redirect():
18
+ return RedirectResponse("/html", status_code=302)
19
+
20
+@app.get("/plain")
21
+def get_plain():
22
+ return PlainTextResponse("Hello, world!")
23
+
24
+def run_server():
25
+ app.run("127.0.0.1:8082")
26
+
27
+if __name__ == "__main__":
28
+ t = threading.Thread(target=run_server, daemon=True)
29
+ t.start()
30
+ time.sleep(1)
31
+
32
+ # JSON
33
+ req = urllib.request.urlopen("http://127.0.0.1:8082/json")
34
+ print("JSON:", req.read().decode(), "Status:", req.status, "Headers:", req.headers.get("content-type"))
35
+
36
+ # HTML
37
+ req = urllib.request.urlopen("http://127.0.0.1:8082/html")
38
+ print("HTML:", req.read().decode(), "Status:", req.status, "Headers:", req.headers.get("content-type"))
39
+
40
+ # Plain
41
+ req = urllib.request.urlopen("http://127.0.0.1:8082/plain")
42
+ print("Plain:", req.read().decode(), "Status:", req.status, "Headers:", req.headers.get("content-type"))
43
+
44
+ # Redirect
45
+ try:
46
+ req = urllib.request.urlopen("http://127.0.0.1:8082/redirect")
47
+ print("Redirect reached:", req.read().decode(), req.status)
48
+ except Exception as e:
49
+ print("Redirect error:", e)
0
50
▼tests/test_sse.py
+12-0
@@ -0,0 +1,12 @@
1
+import asyncio
2
+from justapi import JustAPIApp, TokenStreamResponse
3
+
4
+app = JustAPIApp()
5
+
6
+@app.get("/stream")
7
+async def stream():
8
+ async def gen():
9
+ for i in range(5):
10
+ yield f"data: {i}\n\n"
11
+ await asyncio.sleep(0.1)
12
+ return TokenStreamResponse(gen())
0
13