17026801254

Committed 17 Aug 2025 10:46PM UTC coverage: 67.777% (-31.5%) from 99.312%

Build # 17026801254

Build Type

Pull #7

github

Committed by

web-flow

Commit Message

Merge 6d13f892e into e594bd672

Pull Request Pull Request #7: WIP: Python SDK

Run Details

130 of 215 branches covered (60.47%)

Branch coverage included in aggregate %.

749 of 1074 new or added lines in 20 files covered. (69.74%)

121 existing lines in 7 files now uncovered.

989 of 1436 relevant lines covered (68.87%)

0.69 hits per line

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

72.73

/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,
    HTTP_400_BAD_REQUEST,
    HTTP_415_UNSUPPORTED_MEDIA_TYPE,
    HTTP_408_REQUEST_TIMEOUT,
    HTTP_500_INTERNAL_SERVER_ERROR,
    HTTP_413_REQUEST_ENTITY_TOO_LARGE,
)
import base64
import time

from .models import (
    ConvertResponse,
    ErrorResponse,
)
from .sdk import MDConverter
from .sdk.exceptions import (
    ConversionError,
    InvalidInputError,
    NetworkError,
    TimeoutError,
    FileSizeError,
    UnsupportedFormatError,
)
from .core.config import Settings
from .detection import ContentTypeDetector


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

    @post("")
    async def convert_unified(
        self,
        request: Request,
        md_converter: MDConverter,
        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 SDK for conversion based on input type
            if input_data.get("url"):
                result = await md_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 InvalidInputError("Invalid base64 content")
                else:
                    content = input_data["content"]

                result = await md_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 md_converter.convert_text(input_data["text"], mime_type)
            else:
                raise InvalidInputError(
                    "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 (
            InvalidInputError,
            NetworkError,
            TimeoutError,
            FileSizeError,
            UnsupportedFormatError,
            ConversionError,
        ) as e:
            return self._handle_sdk_error(e)
        except ValueError as e:
            self._handle_value_error(str(e))
        except Exception as e:
            self._handle_generic_error(str(e))

    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_sdk_error(self, error) -> Response[ErrorResponse]:
        """Handle SDK exceptions and map to HTTP responses"""
        if isinstance(error, InvalidInputError):
            error_response = ErrorResponse.create_error(
                code="INVALID_INPUT",
                message=str(error),
                details=error.details,
                suggestions=["Check input format", "Verify request structure"],
            )
            status_code = HTTP_400_BAD_REQUEST

        elif isinstance(error, FileSizeError):
            error_response = ErrorResponse.create_error(
                code="FILE_TOO_LARGE",
                message=str(error),
                details=error.details,
                suggestions=["Use a smaller file", "Check size limits at /formats"],
            )
            status_code = HTTP_413_REQUEST_ENTITY_TOO_LARGE

        elif isinstance(error, UnsupportedFormatError):
            error_response = ErrorResponse.create_error(
                code="UNSUPPORTED_FORMAT",
                message=str(error),
                details=error.details,
                suggestions=["Check supported formats at /formats"],
            )
            status_code = HTTP_415_UNSUPPORTED_MEDIA_TYPE

        elif isinstance(error, TimeoutError):
            error_response = ErrorResponse.create_error(
                code="TIMEOUT",
                message=str(error),
                details=error.details,
                suggestions=["Try with a smaller file", "Increase timeout in options"],
            )
            status_code = HTTP_408_REQUEST_TIMEOUT

        elif isinstance(error, NetworkError):
            error_response = ErrorResponse.create_error(
                code="NETWORK_ERROR",
                message=str(error),
                details=error.details,
                suggestions=["Check URL accessibility", "Verify network connectivity"],
            )
            status_code = HTTP_400_BAD_REQUEST

        else:  # ConversionError or generic
            error_response = ErrorResponse.create_error(
                code="CONVERSION_FAILED",
                message=str(error),
                details=getattr(error, "details", {}),
                suggestions=["Check input format", "Contact support if issue persists"],
            )
            status_code = HTTP_500_INTERNAL_SERVER_ERROR

        raise HTTPException(status_code=status_code, detail=error_response.model_dump())

    def _handle_value_error(self, error_msg: str) -> None:
        """Handle ValueError with specific error type detection"""
        error_mappings = [
            (
                ["size", "exceeds"],
                "FILE_TOO_LARGE",
                HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                ["Use a smaller file", "Check size limits at /formats"],
            ),
            (
                ["not allowed", "blocked"],
                "INVALID_URL",
                HTTP_400_BAD_REQUEST,
                ["Use a public URL", "Avoid private IP addresses"],
            ),
            (
                ["content type mismatch"],
                "INVALID_CONTENT",
                HTTP_400_BAD_REQUEST,
                ["Ensure file matches declared content type"],
            ),
        ]

        for keywords, code, status_code, suggestions in error_mappings:
            if any(keyword in error_msg.lower() for keyword in keywords):
                error_response = ErrorResponse.create_error(
                    code=code, message=error_msg, suggestions=suggestions
                )
                raise HTTPException(
                    status_code=status_code, detail=error_response.model_dump()
                )

        # Default ValueError handling
        error_response = ErrorResponse.create_error(
            code="INVALID_INPUT",
            message=error_msg,
            suggestions=["Check input format", "Verify JSON structure"],
        )
        raise HTTPException(
            status_code=HTTP_400_BAD_REQUEST, detail=error_response.model_dump()
        )

    def _handle_generic_error(self, error_msg: str, format_type: str = None) -> None:
        """Handle generic exceptions"""
        if "unsupported" in error_msg.lower():
            error_response = ErrorResponse.create_error(
                code="UNSUPPORTED_FORMAT",
                message=error_msg,
                details={"detected_format": format_type} if format_type else None,
                suggestions=["Check supported formats at /formats"],
            )
            raise HTTPException(
                status_code=HTTP_415_UNSUPPORTED_MEDIA_TYPE,
                detail=error_response.model_dump(),
            )

        error_response = ErrorResponse.create_error(
            code="CONVERSION_FAILED", message=f"Conversion failed: {error_msg}"
        )
        raise HTTPException(
            status_code=HTTP_500_INTERNAL_SERVER_ERROR,
            detail=error_response.model_dump(),
        )

    def _calculate_source_size(
        self, input_type: str, content_data: dict, request_data: dict
    ) -> int:
        """Calculate source content size (backward compatibility method)"""
        if input_type == "json_url":
            return len(request_data.get("url", "").encode("utf-8"))
        elif input_type in ["json_text", "json_text_typed"]:
            return len(request_data.get("text", "").encode("utf-8"))
        elif input_type == "json_content":
            try:
                return len(base64.b64decode(request_data.get("content", "")))
            except Exception:
                return 0
        elif content_data and "content" in content_data:
            return len(content_data["content"])
        return 0

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	HTTP_400_BAD_REQUEST,
8	HTTP_415_UNSUPPORTED_MEDIA_TYPE,
9	HTTP_408_REQUEST_TIMEOUT,
10	HTTP_500_INTERNAL_SERVER_ERROR,
11	HTTP_413_REQUEST_ENTITY_TOO_LARGE,
12	)
13	import base64	1✔
14	import time	1✔
15
16	from .models import (	1✔
17	ConvertResponse,
18	ErrorResponse,
19	)
20	from .sdk import MDConverter	1✔
21	from .sdk.exceptions import (	1✔
22	ConversionError,
23	InvalidInputError,
24	NetworkError,
25	TimeoutError,
26	FileSizeError,
27	UnsupportedFormatError,
28	)
29	from .core.config import Settings	1✔
30	from .detection import ContentTypeDetector	1✔
31
32
33	class ConvertController(Controller):	1✔
34	path = "/convert"	1✔
35
36	@post("")	1✔
37	async def convert_unified(	1✔
38	self,
39	request: Request,
40	md_converter: MDConverter,
41	settings: Settings,
42	) -> Response[Union[ConvertResponse, ErrorResponse]]:
43	"""Unified conversion endpoint that handles all input types"""
44	start_time = time.time()	1✔
45
46	try:	1✔
47	# Parse request to determine input type and data
48	input_data = await self._parse_request(request)	1✔
49
50	# Use SDK for conversion based on input type
51	if input_data.get("url"):	1✔
52	result = await md_converter.convert_url(	1✔
53	input_data["url"], js_rendering=input_data.get("js_rendering")
54	)
55	elif input_data.get("content"):	1✔
56	# Decode base64 content if needed
57	if isinstance(input_data["content"], str):	1✔
58	try:	1✔
59	content = base64.b64decode(input_data["content"])	1✔
NEW 60	except Exception:	×
NEW 61	raise InvalidInputError("Invalid base64 content")	×
62	else:
63	content = input_data["content"]	1✔
64
65	result = await md_converter.convert_content(	1✔
66	content, filename=input_data.get("filename")
67	)
68	elif input_data.get("text"):	1✔
69	# Determine MIME type: if specified use it, otherwise use markdown for backward compatibility
70	mime_type = input_data.get("mime_type", "text/markdown")	1✔
71	result = await md_converter.convert_text(input_data["text"], mime_type)	1✔
72	else:
73	raise InvalidInputError(	1✔
74	"No valid input provided (url, content, or text)"
75	)
76
77	# Convert SDK result to API response format
78	response = self._create_success_response_from_sdk(result, start_time)	1✔
79	return Response(response, status_code=HTTP_200_OK)	1✔
80
81	except (	1✔
82	InvalidInputError,
83	NetworkError,
84	TimeoutError,
85	FileSizeError,
86	UnsupportedFormatError,
87	ConversionError,
88	) as e:
89	return self._handle_sdk_error(e)	1✔
90	except ValueError as e:	1✔
91	self._handle_value_error(str(e))	1✔
UNCOV 92	except Exception as e:	×
NEW 93	self._handle_generic_error(str(e))	×
94
95	async def _parse_request(self, request: Request) -> dict:	1✔
96	"""Parse request to extract conversion input data"""
97	content_type = request.headers.get("content-type", "")	1✔
98
99	# JSON request
100	if "application/json" in content_type:	1✔
101	try:	1✔
102	json_data = await request.json()	1✔
103
104	# Extract options if present
105	options = json_data.get("options", {})	1✔
106
107	# Add options to the data for SDK consumption
108	result = json_data.copy()	1✔
109	if options:	1✔
110	result.update(options)	1✔
111
112	return result	1✔
113	except Exception:	1✔
114	raise ValueError("Invalid JSON in request body")	1✔
115
116	# Multipart form request
117	elif "multipart/form-data" in content_type:	1✔
118	try:	1✔
119	form_data = await request.form()	1✔
120	if "file" not in form_data:	1✔
121	raise ValueError(	×
122	"File parameter 'file' is required for multipart uploads"
123	)
124
125	file = form_data["file"]	1✔
126	content = await file.read()	1✔
127
128	return {"content": content, "filename": file.filename}	1✔
129
130	except ValueError:	1✔
UNCOV 131	raise	×
132	except Exception as e:	1✔
133	raise ValueError(f"Failed to process multipart upload: {str(e)}")	1✔
134
135	# Binary upload
136	else:
137	try:	1✔
138	content = await request.body()	1✔
139	return {"content": content}	1✔
140
UNCOV 141	except Exception:	×
UNCOV 142	raise ValueError("Failed to read request body")	×
143
144	def _create_success_response_from_sdk(	1✔
145	self, result, start_time: float
146	) -> ConvertResponse:
147	"""Create a successful conversion response from SDK result"""
148	# Calculate total time (including SDK processing time)
149	total_time_ms = int((time.time() - start_time) * 1000)	1✔
150
151	# Use original API source type mapping for backward compatibility
152	# For URL inputs, use "url" as source_type regardless of detected format
153	if result.metadata.source_type == "url":	1✔
NEW 154	source_type = "url"	×
155	else:
156	source_type = ContentTypeDetector.get_source_type(	1✔
157	result.metadata.detected_format
158	)
159
160	return ConvertResponse.create_success(	1✔
161	markdown=result.markdown,
162	source_type=source_type,
163	source_size=result.metadata.source_size,
164	conversion_time_ms=total_time_ms,
165	detected_format=result.metadata.detected_format,
166	warnings=[],
167	)
168
169	def _handle_sdk_error(self, error) -> Response[ErrorResponse]:	1✔
170	"""Handle SDK exceptions and map to HTTP responses"""
171	if isinstance(error, InvalidInputError):	1✔
172	error_response = ErrorResponse.create_error(	1✔
173	code="INVALID_INPUT",
174	message=str(error),
175	details=error.details,
176	suggestions=["Check input format", "Verify request structure"],
177	)
178	status_code = HTTP_400_BAD_REQUEST	1✔
179
180	elif isinstance(error, FileSizeError):	1✔
NEW 181	error_response = ErrorResponse.create_error(	×
182	code="FILE_TOO_LARGE",
183	message=str(error),
184	details=error.details,
185	suggestions=["Use a smaller file", "Check size limits at /formats"],
186	)
NEW 187	status_code = HTTP_413_REQUEST_ENTITY_TOO_LARGE	×
188
189	elif isinstance(error, UnsupportedFormatError):	1✔
NEW 190	error_response = ErrorResponse.create_error(	×
191	code="UNSUPPORTED_FORMAT",
192	message=str(error),
193	details=error.details,
194	suggestions=["Check supported formats at /formats"],
195	)
NEW 196	status_code = HTTP_415_UNSUPPORTED_MEDIA_TYPE	×
197
198	elif isinstance(error, TimeoutError):	1✔
NEW 199	error_response = ErrorResponse.create_error(	×
200	code="TIMEOUT",
201	message=str(error),
202	details=error.details,
203	suggestions=["Try with a smaller file", "Increase timeout in options"],
204	)
NEW 205	status_code = HTTP_408_REQUEST_TIMEOUT	×
206
207	elif isinstance(error, NetworkError):	1✔
208	error_response = ErrorResponse.create_error(	1✔
209	code="NETWORK_ERROR",
210	message=str(error),
211	details=error.details,
212	suggestions=["Check URL accessibility", "Verify network connectivity"],
213	)
214	status_code = HTTP_400_BAD_REQUEST	1✔
215
216	else: # ConversionError or generic
217	error_response = ErrorResponse.create_error(	1✔
218	code="CONVERSION_FAILED",
219	message=str(error),
220	details=getattr(error, "details", {}),
221	suggestions=["Check input format", "Contact support if issue persists"],
222	)
223	status_code = HTTP_500_INTERNAL_SERVER_ERROR	1✔
224
225	raise HTTPException(status_code=status_code, detail=error_response.model_dump())	1✔
226
227	def _handle_value_error(self, error_msg: str) -> None:	1✔
228	"""Handle ValueError with specific error type detection"""
229	error_mappings = [	1✔
230	(
231	["size", "exceeds"],
232	"FILE_TOO_LARGE",
233	HTTP_413_REQUEST_ENTITY_TOO_LARGE,
234	["Use a smaller file", "Check size limits at /formats"],
235	),
236	(
237	["not allowed", "blocked"],
238	"INVALID_URL",
239	HTTP_400_BAD_REQUEST,
240	["Use a public URL", "Avoid private IP addresses"],
241	),
242	(
243	["content type mismatch"],
244	"INVALID_CONTENT",
245	HTTP_400_BAD_REQUEST,
246	["Ensure file matches declared content type"],
247	),
248	]
249
250	for keywords, code, status_code, suggestions in error_mappings:	1✔
251	if any(keyword in error_msg.lower() for keyword in keywords):	1✔
UNCOV 252	error_response = ErrorResponse.create_error(	×
253	code=code, message=error_msg, suggestions=suggestions
254	)
UNCOV 255	raise HTTPException(	×
256	status_code=status_code, detail=error_response.model_dump()
257	)
258
259	# Default ValueError handling
260	error_response = ErrorResponse.create_error(	1✔
261	code="INVALID_INPUT",
262	message=error_msg,
263	suggestions=["Check input format", "Verify JSON structure"],
264	)
265	raise HTTPException(	1✔
266	status_code=HTTP_400_BAD_REQUEST, detail=error_response.model_dump()
267	)
268
269	def _handle_generic_error(self, error_msg: str, format_type: str = None) -> None:	1✔
270	"""Handle generic exceptions"""
UNCOV 271	if "unsupported" in error_msg.lower():	×
UNCOV 272	error_response = ErrorResponse.create_error(	×
273	code="UNSUPPORTED_FORMAT",
274	message=error_msg,
275	details={"detected_format": format_type} if format_type else None,
276	suggestions=["Check supported formats at /formats"],
277	)
UNCOV 278	raise HTTPException(	×
279	status_code=HTTP_415_UNSUPPORTED_MEDIA_TYPE,
280	detail=error_response.model_dump(),
281	)
282
UNCOV 283	error_response = ErrorResponse.create_error(	×
284	code="CONVERSION_FAILED", message=f"Conversion failed: {error_msg}"
285	)
UNCOV 286	raise HTTPException(	×
287	status_code=HTTP_500_INTERNAL_SERVER_ERROR,
288	detail=error_response.model_dump(),
289	)
290
291	def _calculate_source_size(	1✔
292	self, input_type: str, content_data: dict, request_data: dict
293	) -> int:
294	"""Calculate source content size (backward compatibility method)"""
NEW 295	if input_type == "json_url":	×
NEW 296	return len(request_data.get("url", "").encode("utf-8"))	×
NEW 297	elif input_type in ["json_text", "json_text_typed"]:	×
NEW 298	return len(request_data.get("text", "").encode("utf-8"))	×
NEW 299	elif input_type == "json_content":	×
NEW 300	try:	×
NEW 301	return len(base64.b64decode(request_data.get("content", "")))	×
NEW 302	except Exception:	×
NEW 303	return 0	×
NEW 304	elif content_data and "content" in content_data:	×
NEW 305	return len(content_data["content"])	×
NEW 306	return 0	×

peteretelej / md-server / 17026801254

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