17193697155

Committed 24 Aug 2025 08:59PM UTC coverage: 91.351% (-8.0%) from 99.312%

Build # 17193697155

Build Type

Pull #7

github

Committed by

web-flow

Commit Message

Merge 1e0284d36 into e594bd672

Pull Request Pull Request #7: WIP: Python SDK

Run Details

99 of 109 branches covered (90.83%)

Branch coverage included in aggregate %.

384 of 436 new or added lines in 13 files covered. (88.07%)

8 existing lines in 2 files now uncovered.

746 of 816 relevant lines covered (91.42%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.33

/src/md_server/controllers.py

from typing import Union
from litestar import Controller, post, Request
from litestar.response import Response
from litestar.exceptions import HTTPException
from litestar.status_codes import (
    HTTP_200_OK,
)
import base64
import time

from .models import (
    ConvertResponse,
    ErrorResponse,
)
from .core.converter import DocumentConverter
from .core.validation import ValidationError
from .core.config import Settings
from .core.detection import ContentTypeDetector


class ConvertController(Controller):
    path = "/convert"

    @post("")
    async def convert_unified(
        self,
        request: Request,
        document_converter: DocumentConverter,
        settings: Settings,
    ) -> Response[Union[ConvertResponse, ErrorResponse]]:
        """Unified conversion endpoint that handles all input types"""
        start_time = time.time()

        try:
            # Parse request to determine input type and data
            input_data = await self._parse_request(request)

            # Use core converter for conversion based on input type
            if input_data.get("url"):
                result = await document_converter.convert_url(
                    input_data["url"], js_rendering=input_data.get("js_rendering")
                )
            elif input_data.get("content"):
                # Decode base64 content if needed
                if isinstance(input_data["content"], str):
                    try:
                        content = base64.b64decode(input_data["content"])
                    except Exception:
                        raise ValueError("Invalid base64 content")
                else:
                    content = input_data["content"]

                result = await document_converter.convert_content(
                    content, filename=input_data.get("filename")
                )
            elif input_data.get("text"):
                # Determine MIME type: if specified use it, otherwise use markdown for backward compatibility
                mime_type = input_data.get("mime_type", "text/markdown")
                result = await document_converter.convert_text(
                    input_data["text"], mime_type
                )
            else:
                raise ValueError("No valid input provided (url, content, or text)")

            # Convert SDK result to API response format
            response = self._create_success_response_from_sdk(result, start_time)
            return Response(response, status_code=HTTP_200_OK)

        except ValidationError as e:
            return self._handle_validation_error(e)
        except ValueError as e:
            error_response = ErrorResponse.create_error(
                code="INVALID_INPUT",
                message=str(e),
                suggestions=["Check input format", "Verify JSON structure"],
            )
            raise HTTPException(status_code=400, detail=error_response.model_dump())
        except Exception as e:
            error_response = ErrorResponse.create_error(
                code="CONVERSION_FAILED",
                message=f"Conversion failed: {str(e)}",
                suggestions=["Check input format", "Contact support if issue persists"],
            )
            raise HTTPException(status_code=500, detail=error_response.model_dump())

    async def _parse_request(self, request: Request) -> dict:
        """Parse request to extract conversion input data"""
        content_type = request.headers.get("content-type", "")

        # JSON request
        if "application/json" in content_type:
            try:
                json_data = await request.json()

                # Extract options if present
                options = json_data.get("options", {})

                # Add options to the data for SDK consumption
                result = json_data.copy()
                if options:
                    result.update(options)

                return result
            except Exception:
                raise ValueError("Invalid JSON in request body")

        # Multipart form request
        elif "multipart/form-data" in content_type:
            try:
                form_data = await request.form()
                if "file" not in form_data:
                    raise ValueError(
                        "File parameter 'file' is required for multipart uploads"
                    )

                file = form_data["file"]
                content = await file.read()

                return {"content": content, "filename": file.filename}

            except ValueError:
                raise
            except Exception as e:
                raise ValueError(f"Failed to process multipart upload: {str(e)}")

        # Binary upload
        else:
            try:
                content = await request.body()
                return {"content": content}

            except Exception:
                raise ValueError("Failed to read request body")

    def _create_success_response_from_sdk(
        self, result, start_time: float
    ) -> ConvertResponse:
        """Create a successful conversion response from SDK result"""
        # Calculate total time (including SDK processing time)
        total_time_ms = int((time.time() - start_time) * 1000)

        # Use original API source type mapping for backward compatibility
        # For URL inputs, use "url" as source_type regardless of detected format
        if result.metadata.source_type == "url":
            source_type = "url"
        else:
            source_type = ContentTypeDetector.get_source_type(
                result.metadata.detected_format
            )

        return ConvertResponse.create_success(
            markdown=result.markdown,
            source_type=source_type,
            source_size=result.metadata.source_size,
            conversion_time_ms=total_time_ms,
            detected_format=result.metadata.detected_format,
            warnings=[],
        )

    def _handle_validation_error(
        self, error: ValidationError
    ) -> Response[ErrorResponse]:
        """Handle validation exceptions"""
        error_response = ErrorResponse.create_error(
            code="VALIDATION_ERROR",
            message=str(error),
            details=getattr(error, "details", {}),
        )
        raise HTTPException(status_code=400, detail=error_response.model_dump())

1	from typing import Union	1✔
2	from litestar import Controller, post, Request	1✔
3	from litestar.response import Response	1✔
4	from litestar.exceptions import HTTPException	1✔
5	from litestar.status_codes import (	1✔
6	HTTP_200_OK,
7	)
8	import base64	1✔
9	import time	1✔
10
11	from .models import (	1✔
12	ConvertResponse,
13	ErrorResponse,
14	)
15	from .core.converter import DocumentConverter	1✔
16	from .core.validation import ValidationError	1✔
17	from .core.config import Settings	1✔
18	from .core.detection import ContentTypeDetector	1✔
19
20
21	class ConvertController(Controller):	1✔
22	path = "/convert"	1✔
23
24	@post("")	1✔
25	async def convert_unified(	1✔
26	self,
27	request: Request,
28	document_converter: DocumentConverter,
29	settings: Settings,
30	) -> Response[Union[ConvertResponse, ErrorResponse]]:
31	"""Unified conversion endpoint that handles all input types"""
32	start_time = time.time()	1✔
33
34	try:	1✔
35	# Parse request to determine input type and data
36	input_data = await self._parse_request(request)	1✔
37
38	# Use core converter for conversion based on input type
39	if input_data.get("url"):	1✔
40	result = await document_converter.convert_url(	1✔
41	input_data["url"], js_rendering=input_data.get("js_rendering")
42	)
43	elif input_data.get("content"):	1✔
44	# Decode base64 content if needed
45	if isinstance(input_data["content"], str):	1✔
46	try:	1✔
47	content = base64.b64decode(input_data["content"])	1✔
48	except Exception:	1✔
49	raise ValueError("Invalid base64 content")	1✔
50	else:
51	content = input_data["content"]	1✔
52
53	result = await document_converter.convert_content(	1✔
54	content, filename=input_data.get("filename")
55	)
56	elif input_data.get("text"):	1✔
57	# Determine MIME type: if specified use it, otherwise use markdown for backward compatibility
58	mime_type = input_data.get("mime_type", "text/markdown")	1✔
59	result = await document_converter.convert_text(	1✔
60	input_data["text"], mime_type
61	)
62	else:
63	raise ValueError("No valid input provided (url, content, or text)")	1✔
64
65	# Convert SDK result to API response format
66	response = self._create_success_response_from_sdk(result, start_time)	1✔
67	return Response(response, status_code=HTTP_200_OK)	1✔
68
69	except ValidationError as e:	1✔
NEW 70	return self._handle_validation_error(e)	×
71	except ValueError as e:	1✔
72	error_response = ErrorResponse.create_error(	1✔
73	code="INVALID_INPUT",
74	message=str(e),
75	suggestions=["Check input format", "Verify JSON structure"],
76	)
77	raise HTTPException(status_code=400, detail=error_response.model_dump())	1✔
78	except Exception as e:	1✔
79	error_response = ErrorResponse.create_error(	1✔
80	code="CONVERSION_FAILED",
81	message=f"Conversion failed: {str(e)}",
82	suggestions=["Check input format", "Contact support if issue persists"],
83	)
84	raise HTTPException(status_code=500, detail=error_response.model_dump())	1✔
85
86	async def _parse_request(self, request: Request) -> dict:	1✔
87	"""Parse request to extract conversion input data"""
88	content_type = request.headers.get("content-type", "")	1✔
89
90	# JSON request
91	if "application/json" in content_type:	1✔
92	try:	1✔
93	json_data = await request.json()	1✔
94
95	# Extract options if present
96	options = json_data.get("options", {})	1✔
97
98	# Add options to the data for SDK consumption
99	result = json_data.copy()	1✔
100	if options:	1✔
NEW 101	result.update(options)	×
102
103	return result	1✔
104	except Exception:	1✔
105	raise ValueError("Invalid JSON in request body")	1✔
106
107	# Multipart form request
108	elif "multipart/form-data" in content_type:	1✔
109	try:	1✔
110	form_data = await request.form()	1✔
111	if "file" not in form_data:	1✔
112	raise ValueError(	1✔
113	"File parameter 'file' is required for multipart uploads"
114	)
115
116	file = form_data["file"]	1✔
117	content = await file.read()	1✔
118
119	return {"content": content, "filename": file.filename}	1✔
120
121	except ValueError:	1✔
122	raise	1✔
UNCOV 123	except Exception as e:	×
UNCOV 124	raise ValueError(f"Failed to process multipart upload: {str(e)}")	×
125
126	# Binary upload
127	else:
128	try:	1✔
129	content = await request.body()	1✔
130	return {"content": content}	1✔
131
132	except Exception:	1✔
133	raise ValueError("Failed to read request body")	1✔
134
135	def _create_success_response_from_sdk(	1✔
136	self, result, start_time: float
137	) -> ConvertResponse:
138	"""Create a successful conversion response from SDK result"""
139	# Calculate total time (including SDK processing time)
140	total_time_ms = int((time.time() - start_time) * 1000)	1✔
141
142	# Use original API source type mapping for backward compatibility
143	# For URL inputs, use "url" as source_type regardless of detected format
144	if result.metadata.source_type == "url":	1✔
145	source_type = "url"	1✔
146	else:
147	source_type = ContentTypeDetector.get_source_type(	1✔
148	result.metadata.detected_format
149	)
150
151	return ConvertResponse.create_success(	1✔
152	markdown=result.markdown,
153	source_type=source_type,
154	source_size=result.metadata.source_size,
155	conversion_time_ms=total_time_ms,
156	detected_format=result.metadata.detected_format,
157	warnings=[],
158	)
159
160	def _handle_validation_error(	1✔
161	self, error: ValidationError
162	) -> Response[ErrorResponse]:
163	"""Handle validation exceptions"""
UNCOV 164	error_response = ErrorResponse.create_error(	×
165	code="VALIDATION_ERROR",
166	message=str(error),
167	details=getattr(error, "details", {}),
168	)
NEW 169	raise HTTPException(status_code=400, detail=error_response.model_dump())	×

peteretelej / md-server / 17193697155

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous