22874174928

Committed 09 Mar 2026 08:50PM UTC coverage: 90.659% (-0.01%) from 90.671%

Build # 22874174928

Build Type

Pull #14

github

Committed by

web-flow

Commit Message

Merge 2fc7e4692 into 4c6b47c93

Pull Request Pull Request #14: Refactor method ordering to follow Step-down Rule

Coverage Stats

404 of 434 new or added lines in 10 files covered. (93.09%)

2 existing lines in 2 files now uncovered.

1349 of 1488 relevant lines covered (90.66%)

4.53 hits per line

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

82.93

/src/chunklet/visualizer/visualizer.py

import json
import mimetypes
import traceback
from pathlib import Path
from typing import Callable

import aiofiles

try:
    import uvicorn
    from charset_normalizer import detect
    from fastapi import FastAPI, File, Form, HTTPException, UploadFile
    from fastapi.responses import HTMLResponse
    from fastapi.staticfiles import StaticFiles
except ImportError:  # pragma: no cover
    # Lambda placeholders prevent "None is not callable" errors when imports fail
    # This allows the module to be imported without dependencies, with proper error handling later
    uvicorn = None
    detect = None
    FastAPI = None
    UploadFile = None
    File = lambda x: x  # noqa: E731
    Form = lambda x: x  # noqa: E731
    HTTPException = None
    HTMLResponse = lambda x: x  # noqa: E731
    StaticFiles = None

from chunklet.code_chunker import CodeChunker
from chunklet.common.validation import validate_input
from chunklet.document_chunker import DocumentChunker


class Visualizer:
    """A FastAPI-based web interface for visualizing document and code chunks.

    This server allows users to upload text or code files, processes them with
    Chunklet's `DocumentChunker` or `CodeChunker`, and returns the chunked
    data along with statistics. A minimal frontend interface is served at the
    root endpoint.

    Attributes:
        host (str): Host IP to bind the FastAPI server.
        port (int): Port number to run the server on.
        document_chunker (DocumentChunker): Chunklet document chunker instance.
        code_chunker (CodeChunker): Chunklet code chunker instance.
        app (FastAPI): FastAPI application instance.
    """

    @validate_input
    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 8000,
        token_counter: Callable[[str], int] | None = None,
    ):
        """Initializes the Visualizer server and configures chunkers.

        Args:
            host (str): Host IP to run the server. Defaults to "127.0.0.1".
            port (int): Port number to run the server. Defaults to 8000.
            token_counter (Optional[Callable[[str], int]]): Function to count tokens
                in text/code. Required for chunkers if used with `max_tokens`.
        """
        if FastAPI is None:
            raise ImportError(
                "The 'fastapi' library is not installed. "
                "Please install it with 'pip install fastapi>=0.115.12' or install the visualization extras "
                "with 'pip install 'chunklet-py[visualization]''"
            )

        self.host = host
        self.port = port
        self._token_counter = token_counter

        self.app = FastAPI()

        # Initialize chunkers
        self.document_chunker = DocumentChunker(token_counter=token_counter)
        self.code_chunker = CodeChunker(token_counter=token_counter)

        self.static_dir = Path(__file__).parent / "static"

        self.app.mount(
            "/static", StaticFiles(directory=str(self.static_dir)), name="static"
        )

        # API endpoints
        self.app.get("/api/token_counter_status")(self._get_token_counter_status)
        self.app.get("/health")(self._get_health_check)
        self.app.get("/")(self._get_index)
        self.app.post("/api/chunk")(self._chunk_file)

    @property
    def token_counter(self):
        """Get the current token counter function."""
        return self._token_counter

    @token_counter.setter
    @validate_input
    def token_counter(self, value):
        """Set the token counter and update both chunkers."""
        self._token_counter = value
        self.document_chunker.token_counter = value
        self.code_chunker.token_counter = value

    def serve(self):
        """Starts the FastAPI server and prints the server URL."""
        if uvicorn is None:
            raise ImportError(
                "The 'uvicorn' library is not installed. "
                "Please install it with 'pip install uvicorn>=0.34.0' or install the visualization extras "
                "with 'pip install 'chunklet-py[visualization]''"
            )

        print(" =" * 20)
        print("\nTEXT CHUNK VISUALIZER")
        print("= " * 20)
        print(f"URL: http://{self.host}:{self.port}")

        uvicorn.run(self.app, host=self.host, port=self.port)

    # Instance endpoint methods
    def _get_token_counter_status(self):
        return {"token_counter_available": self.token_counter is not None}

    def _get_health_check(self):
        """Health check endpoint for testing."""
        return {"status": "healthy"}

    async def _get_index(self):
        """Serves the main HTML interface for the visualizer.

        Returns:
            HTMLResponse: The content of index.html if exists, else a default heading.
        """
        index_path = self.static_dir / "index.html"
        if index_path.exists():
            async with aiofiles.open(index_path, "r", encoding="utf-8") as f:
                content = await f.read()
                return HTMLResponse(content=content)
        return HTMLResponse(content="<h1>Text Chunk Visualizer</h1>")

    @validate_input
    async def _chunk_file(
        self,
        file: UploadFile = File(...),
        mode: str = Form("document"),
        params: str = Form("{}"),
    ) -> dict:
        """Processes an uploaded file and returns chunked output.

        Args:
            self: The Visualizer instance.
            file (UploadFile): File uploaded by the client.
            mode (str): Determines which chunker to use ("document" or "code").
            params (str): JSON string containing chunking parameters.

        Returns:
            dict: Contains original text, chunked content, and statistics.

        Raises:
            HTTPException: If chunking fails.
        """
        # Parse params JSON and filter out None values
        try:
            chunker_params = json.loads(params)
            chunker_params = {k: v for k, v in chunker_params.items() if v is not None}
        except (json.JSONDecodeError, TypeError):
            raise HTTPException(
                400, f"Invalid chunking parameters JSON: {params}"
            ) from None

        # Use Python mimetypes instead of browser content_type
        mimetype, _ = mimetypes.guess_type(file.filename or "")
        if not mimetype or not mimetype.startswith("text/"):
            raise HTTPException(400, "Only text files are supported.")

        if detect is None:
            raise HTTPException(
                400,
                "charset-normalizer library is not available. Please install visualization dependencies."
                "with 'pip install 'chunklet-py[visualization]''",
            )

        content = await file.read()
        encoding = detect(content).get("encoding", "utf-8")
        text = content.decode(encoding, errors="ignore")
        chunker = self.code_chunker if mode == "code" else self.document_chunker

        try:
            chunks = [
                dict(chunk) for chunk in chunker.chunk_text(text, **chunker_params)
            ]

            return {
                "text": text,
                "chunks": chunks,
                "stats": {
                    "text_length": len(text),
                    "chunk_count": len(chunks),
                    "mode": mode,
                },
            }

        except Exception as e:
            traceback.print_exc()
            raise HTTPException(
                500,
                f"Chunking failed. Please check the server terminal for specific error details. ({str(e)})",
            ) from None


if __name__ == "__main__":  # pragma: no cover
    visualizer = Visualizer()
    visualizer.serve()

1	import json	5✔
2	import mimetypes	5✔
3	import traceback	5✔
4	from pathlib import Path	5✔
5	from typing import Callable	5✔
6
7	import aiofiles	5✔
8
9	try:	5✔
10	import uvicorn	5✔
11	from charset_normalizer import detect	5✔
12	from fastapi import FastAPI, File, Form, HTTPException, UploadFile	5✔
13	from fastapi.responses import HTMLResponse	5✔
14	from fastapi.staticfiles import StaticFiles	5✔
15	except ImportError: # pragma: no cover
16	# Lambda placeholders prevent "None is not callable" errors when imports fail
17	# This allows the module to be imported without dependencies, with proper error handling later
18	uvicorn = None
19	detect = None
20	FastAPI = None
21	UploadFile = None
22	File = lambda x: x # noqa: E731
23	Form = lambda x: x # noqa: E731
24	HTTPException = None
25	HTMLResponse = lambda x: x # noqa: E731
26	StaticFiles = None
27
28	from chunklet.code_chunker import CodeChunker	5✔
29	from chunklet.common.validation import validate_input	5✔
30	from chunklet.document_chunker import DocumentChunker	5✔
31
32
33	class Visualizer:	5✔
34	"""A FastAPI-based web interface for visualizing document and code chunks.
35
36	This server allows users to upload text or code files, processes them with
37	Chunklet's `DocumentChunker` or `CodeChunker`, and returns the chunked
38	data along with statistics. A minimal frontend interface is served at the
39	root endpoint.
40
41	Attributes:
42	host (str): Host IP to bind the FastAPI server.
43	port (int): Port number to run the server on.
44	document_chunker (DocumentChunker): Chunklet document chunker instance.
45	code_chunker (CodeChunker): Chunklet code chunker instance.
46	app (FastAPI): FastAPI application instance.
47	"""
48
49	@validate_input	5✔
50	def __init__(	5✔
51	self,
52	host: str = "127.0.0.1",
53	port: int = 8000,
54	token_counter: Callable[[str], int] \| None = None,
55	):
56	"""Initializes the Visualizer server and configures chunkers.
57
58	Args:
59	host (str): Host IP to run the server. Defaults to "127.0.0.1".
60	port (int): Port number to run the server. Defaults to 8000.
61	token_counter (Optional[Callable[[str], int]]): Function to count tokens
62	in text/code. Required for chunkers if used with `max_tokens`.
63	"""
64	if FastAPI is None:	5✔
65	raise ImportError(	×
66	"The 'fastapi' library is not installed. "
67	"Please install it with 'pip install fastapi>=0.115.12' or install the visualization extras "
68	"with 'pip install 'chunklet-py[visualization]''"
69	)
70
71	self.host = host	5✔
72	self.port = port	5✔
73	self._token_counter = token_counter	5✔
74
75	self.app = FastAPI()	5✔
76
77	# Initialize chunkers
78	self.document_chunker = DocumentChunker(token_counter=token_counter)	5✔
79	self.code_chunker = CodeChunker(token_counter=token_counter)	5✔
80
81	self.static_dir = Path(__file__).parent / "static"	5✔
82
83	self.app.mount(	5✔
84	"/static", StaticFiles(directory=str(self.static_dir)), name="static"
85	)
86
87	# API endpoints
88	self.app.get("/api/token_counter_status")(self._get_token_counter_status)	5✔
89	self.app.get("/health")(self._get_health_check)	5✔
90	self.app.get("/")(self._get_index)	5✔
91	self.app.post("/api/chunk")(self._chunk_file)	5✔
92
93	@property	5✔
94	def token_counter(self):	5✔
95	"""Get the current token counter function."""
96	return self._token_counter	5✔
97
98	@token_counter.setter	5✔
99	@validate_input	5✔
100	def token_counter(self, value):	5✔
101	"""Set the token counter and update both chunkers."""
102	self._token_counter = value	5✔
103	self.document_chunker.token_counter = value	5✔
104	self.code_chunker.token_counter = value	5✔
105
106	def serve(self):	5✔
107	"""Starts the FastAPI server and prints the server URL."""
108	if uvicorn is None:	5✔
NEW 109	raise ImportError(	×
110	"The 'uvicorn' library is not installed. "
111	"Please install it with 'pip install uvicorn>=0.34.0' or install the visualization extras "
112	"with 'pip install 'chunklet-py[visualization]''"
113	)
114
115	print(" =" * 20)	5✔
116	print("\nTEXT CHUNK VISUALIZER")	5✔
117	print("= " * 20)	5✔
118	print(f"URL: http://{self.host}:{self.port}")	5✔
119
120	uvicorn.run(self.app, host=self.host, port=self.port)	5✔
121
122	# Instance endpoint methods
123	def _get_token_counter_status(self):	5✔
124	return {"token_counter_available": self.token_counter is not None}	5✔
125
126	def _get_health_check(self):	5✔
127	"""Health check endpoint for testing."""
128	return {"status": "healthy"}	5✔
129
130	async def _get_index(self):	5✔
131	"""Serves the main HTML interface for the visualizer.
132
133	Returns:
134	HTMLResponse: The content of index.html if exists, else a default heading.
135	"""
136	index_path = self.static_dir / "index.html"	×
137	if index_path.exists():	×
138	async with aiofiles.open(index_path, "r", encoding="utf-8") as f:	×
139	content = await f.read()	×
140	return HTMLResponse(content=content)	×
141	return HTMLResponse(content="<h1>Text Chunk Visualizer</h1>")	×
142
143	@validate_input	5✔
144	async def _chunk_file(	5✔
145	self,
146	file: UploadFile = File(...),
147	mode: str = Form("document"),
148	params: str = Form("{}"),
149	) -> dict:
150	"""Processes an uploaded file and returns chunked output.
151
152	Args:
153	self: The Visualizer instance.
154	file (UploadFile): File uploaded by the client.
155	mode (str): Determines which chunker to use ("document" or "code").
156	params (str): JSON string containing chunking parameters.
157
158	Returns:
159	dict: Contains original text, chunked content, and statistics.
160
161	Raises:
162	HTTPException: If chunking fails.
163	"""
164	# Parse params JSON and filter out None values
165	try:	5✔
166	chunker_params = json.loads(params)	5✔
167	chunker_params = {k: v for k, v in chunker_params.items() if v is not None}	5✔
168	except (json.JSONDecodeError, TypeError):	×
169	raise HTTPException(	×
170	400, f"Invalid chunking parameters JSON: {params}"
171	) from None
172
173	# Use Python mimetypes instead of browser content_type
174	mimetype, _ = mimetypes.guess_type(file.filename or "")	5✔
175	if not mimetype or not mimetype.startswith("text/"):	5✔
176	raise HTTPException(400, "Only text files are supported.")	5✔
177
178	if detect is None:	5✔
179	raise HTTPException(	×
180	400,
181	"charset-normalizer library is not available. Please install visualization dependencies."
182	"with 'pip install 'chunklet-py[visualization]''",
183	)
184
185	content = await file.read()	5✔
186	encoding = detect(content).get("encoding", "utf-8")	5✔
187	text = content.decode(encoding, errors="ignore")	5✔
188	chunker = self.code_chunker if mode == "code" else self.document_chunker	5✔
189
190	try:	5✔
191	chunks = [	5✔
192	dict(chunk) for chunk in chunker.chunk_text(text, **chunker_params)
193	]
194
195	return {	5✔
196	"text": text,
197	"chunks": chunks,
198	"stats": {
199	"text_length": len(text),
200	"chunk_count": len(chunks),
201	"mode": mode,
202	},
203	}
204
205	except Exception as e:	×
206	traceback.print_exc()	×
207	raise HTTPException(	×
208	500,
209	f"Chunking failed. Please check the server terminal for specific error details. ({str(e)})",
210	) from None
211
212
213	if __name__ == "__main__": # pragma: no cover
214	visualizer = Visualizer()
215	visualizer.serve()

speedyk-005 / chunklet-py / 22874174928

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