Skip to content

chore: clean code #90

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
j03-dev merged 6 commits into from
Aug 3, 2026
Merged

chore: clean code #90

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4802d12
chore: first try
j03-dev Aug 2, 2026
f5101ee
chore: pass middleware instead of router
j03-dev Aug 2, 2026
3035101
chore: make test sync
j03-dev Aug 2, 2026
5d354a9
chore: slice middlewares
j03-dev Aug 2, 2026
f150d06
chore: remove transmute in find method
j03-dev Aug 2, 2026
14cca46
chore: improve code as possible
j03-dev Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 76 additions & 93 deletions src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
#![allow(unused_variables, non_snake_case)]

use std::net::SocketAddr;
use std::ops::Deref;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

Expand All @@ -25,6 +24,8 @@ use routing::*;
use status::Status;
use templating::Template;

use crate::middleware::Middleware;

mod cors;
#[macro_use]
mod exceptions;
Expand All @@ -44,21 +45,43 @@ pyo3_stub_gen::export_verbatim!("oxapy", "from typing_extensions import Self");
pyo3_stub_gen::define_stub_info_gatherer!(stub_info);

struct ProcessRequest {
wrapper: Option<Arc<Py<PyAny>>>,
router: Option<Arc<Router>>,
match_route: Option<MatchRoute<'static>>,
middlewares: Option<Arc<[Middleware]>>,
request: Arc<Request>,
tx: Sender<Response>,
response_sender: Sender<Response>,
wrapper: Option<Arc<Py<PyAny>>>,
}

#[derive(Clone)]
struct RequestContext {
struct Context {
app_data: Option<Arc<Py<PyAny>>>,
wrapper: Option<Arc<Py<PyAny>>>,
channel_capacity: usize,
routers: Vec<Arc<Router>>,
request_sender: Sender<ProcessRequest>,
routers: Vec<Arc<Router>>,
template: Option<Arc<Template>>,
wrapper: Option<Arc<Py<PyAny>>>,
}

struct ShutDownSignal {
rx: Receiver<()>,
}

impl ShutDownSignal {
fn new() -> PyResult<Self> {
let running = Arc::new(AtomicBool::new(true));
let (tx, rx) = channel::<()>(1);
ctrlc::set_handler(move || {
println!("\nShutting Down...");
running.store(false, Ordering::SeqCst);
let _ = block_on(tx.send(()), None);
})
.into_py_exception()?;
Ok(Self { rx })
}

async fn wait(&mut self) {
self.rx.recv().await;
}
}

/// HTTP Server for handling web requests.
Expand Down Expand Up @@ -430,33 +453,25 @@ impl HttpServer {

impl HttpServer {
async fn run_server(&self) -> PyResult<()> {
let (listener, shutdown) = self.setup_serve().await?;
let (ctx, rx) = self.create_request_context();
self.spawn_connection_handler(listener, Arc::new(ctx)).await;
self.process_requests(shutdown, rx).await
}

async fn setup_serve(&self) -> PyResult<(TcpListener, ShutDownSignal)> {
let listener = TcpListener::bind(self.addr).await?;
println!("Listening on {}", self.addr);
let shutdown = ShutDownSignal::new()?;
Ok((listener, shutdown))
}

fn create_request_context(&self) -> (RequestContext, Receiver<ProcessRequest>) {
let (tx, rx) = channel::<ProcessRequest>(self.channel_capacity);
let ctx = RequestContext {
let (request_sender, request_receiver) = channel::<ProcessRequest>(self.channel_capacity);
let ctx = Context {
app_data: self.app_data.clone(),
wrapper: self.wrapper.clone(),
channel_capacity: self.channel_capacity,
request_sender,
routers: self.routers.clone(),
request_sender: tx,
template: self.template.clone(),
wrapper: self.wrapper.clone(),
};
(ctx, rx)

self.spawn_connection_handler(listener, Arc::new(ctx)).await;
self.process_requests(shutdown, request_receiver).await
}

async fn spawn_connection_handler(&self, listener: TcpListener, ctx: Arc<RequestContext>) {
async fn spawn_connection_handler(&self, listener: TcpListener, ctx: Arc<Context>) {
let running = self.running.clone();
let max_connection = self.max_connections.clone();
tokio::spawn(async move {
Expand All @@ -473,7 +488,7 @@ impl HttpServer {

fn spawn_request_handler(
io: hyper_util::rt::TokioIo<TcpStream>,
ctx: Arc<RequestContext>,
ctx: Arc<Context>,
_permit: tokio::sync::OwnedSemaphorePermit,
) {
tokio::spawn(async move {
Expand All @@ -487,13 +502,14 @@ impl HttpServer {
hyper::service::service_fn(move |req| {
let ctx = ctx.clone();
async move {
let request = RequestBuilder::new(req)
RequestBuilder::new(req)
.with_app_data(&ctx.app_data)
.with_template(&ctx.template)
.build()
.await
.unwrap();
request.process(ctx).await
.unwrap()
.process(ctx)
.await
Comment on lines +505 to +512

@coderabbitai coderabbitai Bot Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win

Remove the .unwrap() on RequestBuilder::build(); a remote client can panic the connection task.

build() returns PyResult<Request> and fails on conditions that a client controls. src/request.rs lines 350-381 show two failure paths: self.req.collect().await.into_py_exception()? fails when the body read fails, for example on an aborted upload, and parse_multipart(...).into_py_exception()? fails on a malformed multipart/form-data body. Each failure now panics inside the hyper service task instead of producing a response.

Return a 500 response for the error case.

🛡️ Proposed fix 
                 hyper::service::service_fn(move |req| {
                     let ctx = ctx.clone();
                     async move {
-                        RequestBuilder::new(req)
+                        let request = RequestBuilder::new(req)
                             .with_app_data(&ctx.app_data)
                             .with_template(&ctx.template)
                             .build()
-                            .await
-                            .unwrap()
-                            .process(ctx)
-                            .await
+                            .await;
+                        match request {
+                            Ok(request) => request.process(ctx).await,
+                            Err(_) => Response::from(Status::INTERNAL_SERVER_ERROR).try_into(),
+                        }
                     }
                 }),
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
RequestBuilder::new(req)
.with_app_data(&ctx.app_data)
.with_template(&ctx.template)
.build()
.await
.unwrap();
request.process(ctx).await
.unwrap()
.process(ctx)
.await
let request = RequestBuilder::new(req)
.with_app_data(&ctx.app_data)
.with_template(&ctx.template)
.build()
.await;
match request {
Ok(request) => request.process(ctx).await,
Err(_) => Response::from(Status::INTERNAL_SERVER_ERROR).try_into(),
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/lib.rs` around lines 505 - 512, Handle the `RequestBuilder::build()`
error in the request processing chain instead of calling `.unwrap()`, returning
a 500 response when body collection or multipart parsing fails. Preserve the
existing `.process(ctx).await` flow for successfully built requests.

}
}),
)
Expand All @@ -505,90 +521,57 @@ impl HttpServer {
async fn process_requests(
&self,
mut shutdown: ShutDownSignal,
mut rx: Receiver<ProcessRequest>,
mut request_receiver: Receiver<ProcessRequest>,
) -> PyResult<()> {
loop {
tokio::select! {
Some(req) = rx.recv() => self.handle_request(req).await?,
Some(pr) = request_receiver.recv() => {
let response = call_python_handler(&pr.middlewares, &pr.match_route, &pr.request, self.is_async)
.await
.unwrap_or_else(Response::from)
.call_wrapper(&pr);
let _ = pr.response_sender.send(response).await;
},
_ = shutdown.wait() => break,
}
}
Ok(())
}
Comment on lines 521 to 539

@coderabbitai coderabbitai Bot Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Inspect the previous dispatch loop and any task spawning around request processing.
rg -n -C6 'process_requests|tokio::spawn|call_python_handler' src/lib.rs
git log --oneline -5 -- src/lib.rs

Repository: j03-dev/oxapy

Length of output: 2983

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== struct and relevant fields =="
rg -n -C4 'struct .*|is_async|max_connections|channel_capacity|request_sender|ProcessRequest|send\(request' src/lib.rs

echo
echo "== process_requests surrounding code =="
sed -n '430,545p' src/lib.rs | nl -ba -v430

echo
echo "== call_python_handler implementation =="
sed -n '542,620p' src/lib.rs | nl -ba -v542

Repository: j03-dev/oxapy

Length of output: 6546

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== call_python_handler implementation =="
sed -n '542,585p' src/lib.rs

echo
echo "== request sending locations =="
rg -n -C5 'request_sender|ProcessRequest|response_receiver|recv\\(' src/lib.rs

echo
echo "== deterministic model: one process_requests task awaiting Python handler keeps loop at one in-flight request =="
python3 - <<'PY'
events = []
pr_channel = []
permitted_connections = 3
process_requests_in_flight = 0

def event(s): events.append(s)
def spawn_connection_accept():
    pr_channel.append('req')
    event("accept_and_send_request")

def process_request_batch():
    global process_requests_in_flight
    if pr_channel:
        pr_channel.pop(0)
        event("process_requests_recieve_and_await_handler_started")
        process_requests_in_flight += 1
    # one await point; while handler awaits, no next recv occurs.
    event("handler_awaits_returns")
    process_requests_in_flight -= 1
    event("process_requests_responds_recieve_next_request")

spawn_connection_accept()
spawn_connection_accept()
spawn_connection_accept()
process_request_batch()
process_request_batch()
process_request_batch()

print("requests_sent:", len([e for e in events if e == "accept_and_send_request"]))
print("requests_in_channel:", len(pr_channel))
print("process_requests_in_flight:", process_requests_in_flight)
print("events:", events)
PY

Repository: j03-dev/oxapy

Length of output: 1770

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== request sending and response waiting locations =="
rg -n -C5 'request_sender|ProcessRequest|response_receiver|recv\\(' src/lib.rs || true

echo
echo "== pyo3 detach helpers in file =="
rg -n -C4 'py.detach|block_on|future_into_py|Python::attach|into_future|allow_threads' src/lib.rs

echo
echo "== imports / no_std context hints =="
sed -n '1,80p' src/lib.rs

Repository: j03-dev/oxapy

Length of output: 4602

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== all ProcessRequest/request_sender usages =="
grep -R -n "ProcessRequest\|request_sender" src || true

echo
echo "== HttpServer request handlers and channel buffering =="

# print matching context around likely request entry points and Channel/Sender fields, line by line
python3 - <<'PY'
from pathlib import Path
p = Path('src/lib.rs')
text = p.read_text().splitlines()
targets = ['request_sender', 'ProcessRequest', 'ctx.request_sender.send', 'response_receiver', 'recv(']
seen = set()
for i,l in enumerate(text, start=1):
    if any(t in l for t in targets):
        start=max(1,i-20); end=min(len(text),i+35)
        key=(start,end)
        if key in seen: continue
        seen.add(key)
        print(f"--- src/lib.rs:{i} ---")
        for n in range(start,end+1):
            print(f"{n:4}: {text[n-1]}")
PY

Repository: j03-dev/oxapy

Length of output: 16060

Spawn each request instead of awaiting the Python handler inline.

process_requests receives from the channel, then awaits call_python_handler(...) before receiving the next request. This makes channel_capacity only a queue limit and caps in-flight request concurrency to 1 for async handlers, while I/O-bound handlers block the only consumer. Spawn per-ProcessRequest, then await its response only after sending it or with per-request capacity.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/lib.rs` around lines 521 - 539, Update process_requests so each received
ProcessRequest is handled in a spawned task rather than awaiting
call_python_handler inline, allowing the receiver loop to continue accepting
requests concurrently. Move response construction and response_sender.send into
the per-request task, while preserving shutdown handling and existing
error-to-Response conversion.


async fn handle_request(&self, req: ProcessRequest) -> PyResult<()> {
let response =
call_python_handler(&req.router, &req.match_route, &req.request, self.is_async)
.await
.unwrap_or_else(Response::from)
.call_wrapper(&req);
let _ = req.tx.send(response).await;
Ok(())
}
}

struct ShutDownSignal {
rx: Receiver<()>,
}

impl ShutDownSignal {
fn new() -> PyResult<Self> {
let running = Arc::new(AtomicBool::new(true));
let (tx, rx) = channel::<()>(1);
ctrlc::set_handler(move || {
println!("\nShutting Down...");
running.store(false, Ordering::SeqCst);
let _ = block_on(tx.send(()), None);
})
.into_py_exception()?;
Ok(Self { rx })
}

async fn wait(&mut self) {
self.rx.recv().await;
}
}

async fn call_python_handler<'l>(
router: &Option<Arc<Router>>,
middlewares: &Option<Arc<[Middleware]>>,
match_route: &Option<MatchRoute<'l>>,
request: &Request,
is_async: bool,
) -> PyResult<Response> {
match (match_route, router) {
(Some(route), Some(router)) => {
let mut result = execute_route_handler(route, router, request)?;
if is_async {
result = Python::attach(|py| into_future(result.into_bound(py)))?.await?;
if let Some(match_route) = match_route {
let mut result = Python::attach(|py| {
let route = match_route.value;
let params = &match_route.params;
let kwargs = build_route_params(py, params)?;

match middlewares {
Some(middlewares) => MiddlewareChain::execute(
py,
middlewares,
route.sequence,
&route.handler,
(request.clone(),),
kwargs,
),
None => route.handler.call(py, (request.clone(),), Some(&kwargs)),
}
Python::attach(|py| into_response::convert_to_response(result, py))
}
_ => Ok(Status::NOT_FOUND.into()),
}
}
})?;

fn execute_route_handler(
match_route: &MatchRoute,
router: &Router,
request: &Request,
) -> PyResult<Py<PyAny>> {
Python::attach(|py| {
let route = match_route.value;
let params = &match_route.params;
let kwargs = build_route_params(py, params)?;
if router.middlewares.is_empty() {
route.handler.call(py, (request.clone(),), Some(&kwargs))
} else {
let chain = MiddlewareChain::new(&router.middlewares);
chain.execute(
py,
route.sequence,
route.handler.deref(),
(request.clone(),),
kwargs.clone(),
)
if is_async {
result = Python::attach(|py| into_future(result.into_bound(py)))?.await?;
}
})

Python::attach(|py| into_response::convert_to_response(result, py))
} else {
Ok(Status::NOT_FOUND.into())
}
}

fn build_route_params<'py>(
Expand Down
28 changes: 14 additions & 14 deletions src/middleware.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,18 +17,12 @@ impl Middleware {
}
}

pub struct MiddlewareChain<'l> {
middlewares: &'l [Middleware],
}

impl<'l> MiddlewareChain<'l> {
pub fn new(middlewares: &'l [Middleware]) -> Self {
Self { middlewares }
}
pub struct MiddlewareChain;

impl MiddlewareChain {
pub fn execute<'py, A>(
&self,
py: Python<'py>,
middlewares: &[Middleware],
route_sequence: usize,
route_handler: &Py<PyAny>,
args: A,
Expand All @@ -37,25 +31,31 @@ impl<'l> MiddlewareChain<'l> {
where
A: PyCallArgs<'py>,
{
let handler = self.build_middleware_chain(py, route_sequence, route_handler, 0)?;
let handler =
Self::build_middleware_chain(py, middlewares, route_sequence, route_handler, 0)?;
handler.call(py, args, Some(&kwargs))
}

fn build_middleware_chain(
&self,
py: Python<'_>,
middlewares: &[Middleware],
route_sequence: usize,
route_handler: &Py<PyAny>,
index: usize,
) -> PyResult<Py<PyAny>> {
let Some(middleware) = self
.middlewares
let Some(middleware) = middlewares
.get(index)
.filter(|m| m.sequence <= route_sequence)
else {
return Ok(route_handler.clone_ref(py));
};
let next = self.build_middleware_chain(py, route_sequence, route_handler, index + 1)?;
let next = Self::build_middleware_chain(
py,
middlewares,
route_sequence,
route_handler,
index + 1,
)?;
let globals = PyDict::new(py);
globals.set_item("middleware", middleware.handler.clone_ref(py))?;
globals.set_item("next", next)?;
Expand Down
Loading
Loading