Coverage for fastblocks / adapters / templates / _language_server.py: 33%

1"""FastBlocks Language Server Protocol implementation.""" 

2 

3import asyncio 

4import typing as t 

5from contextlib import suppress 

6from typing import Any 

7from uuid import UUID 

8 

9from acb.config import Settings 

10from acb.depends import depends 

11 

12from ._syntax_support import FastBlocksSyntaxSupport 

13 

14 

15class LanguageServerSettings(Settings): 

16 """Settings for FastBlocks Language Server.""" 

17 

18 # Required ACB 0.19.0+ metadata 

19 MODULE_ID: UUID = UUID("01937d87-2345-6789-abcd-123456789def") 

20 MODULE_STATUS: str = "stable" 

21 

22 # Server settings 

23 port: int = 7777 

24 host: str = "localhost" 

25 enable_tcp: bool = False 

26 enable_stdio: bool = True 

27 

28 # Feature flags 

29 enable_completions: bool = True 

30 enable_hover: bool = True 

31 enable_diagnostics: bool = True 

32 enable_formatting: bool = True 

33 enable_signature_help: bool = True 

34 

35 # Performance settings 

36 completion_trigger_characters: list[str] = ["[", "|", ".", "("] 

37 signature_trigger_characters: list[str] = ["(", ","] 

38 diagnostic_delay_ms: int = 500 

39 completion_timeout_ms: int = 1000 

40 

41 

42class FastBlocksLanguageServer: 

43 """Language Server Protocol implementation for FastBlocks templates.""" 

44 

45 # Required ACB 0.19.0+ metadata 

46 MODULE_ID: UUID = UUID("01937d87-2345-6789-abcd-123456789def") 

47 MODULE_STATUS: str = "stable" 

48 

49 def __init__(self) -> None: 

50 """Initialize language server.""" 

51 self.settings: LanguageServerSettings | None = None 

52 self.syntax_support: FastBlocksSyntaxSupport | None = None 

53 self._documents: dict[str, str] = {} 

54 self._diagnostics: dict[str, list[dict[str, Any]]] = {} 

55 

56 # Register with ACB 

57 with suppress(Exception): 

58 depends.set(self) 

59 

60 # Initialize syntax support 

61 self.syntax_support = FastBlocksSyntaxSupport() 

62 

63 async def initialize(self, params: dict[str, Any]) -> dict[str, Any]: 

64 """Handle LSP initialize request.""" 

65 if not self.settings: 

66 self.settings = LanguageServerSettings() 

67 

68 capabilities: dict[str, Any] = { 

69 "textDocumentSync": { 

70 "openClose": True, 

71 "change": 1, # Full document sync 

72 "save": {"includeText": True}, 

73 } 

74 } 

75 

76 if self.settings.enable_completions: 

77 capabilities["completionProvider"] = { 

78 "triggerCharacters": self.settings.completion_trigger_characters, 

79 "resolveProvider": True, 

80 } 

81 

82 if self.settings.enable_hover: 

83 capabilities["hoverProvider"] = True 

84 

85 if self.settings.enable_formatting: 

86 capabilities["documentFormattingProvider"] = True 

87 

88 if self.settings.enable_signature_help: 

89 capabilities["signatureHelpProvider"] = { 

90 "triggerCharacters": self.settings.signature_trigger_characters 

91 } 

92 

93 return { 

94 "capabilities": capabilities, 

95 "serverInfo": {"name": "FastBlocks Language Server", "version": "1.0.0"}, 

96 } 

97 

98 async def text_document_did_open(self, params: dict[str, Any]) -> None: 

99 """Handle document open event.""" 

100 doc = params["textDocument"] 

101 uri = doc["uri"] 

102 content = doc["text"] 

103 

104 self._documents[uri] = content 

105 

106 # Run diagnostics 

107 if self.settings and self.settings.enable_diagnostics: 

108 await self._run_diagnostics(uri, content) 

109 

110 async def text_document_did_change(self, params: dict[str, Any]) -> None: 

111 """Handle document change event.""" 

112 uri = params["textDocument"]["uri"] 

113 changes = params["contentChanges"] 

114 

115 # For full document sync 

116 if changes: 

117 self._documents[uri] = changes[0]["text"] 

118 

119 # Delayed diagnostics 

120 if self.settings and self.settings.enable_diagnostics: 

121 await asyncio.sleep(self.settings.diagnostic_delay_ms / 1000) 

122 await self._run_diagnostics(uri, self._documents[uri]) 

123 

124 async def text_document_completion(self, params: dict[str, Any]) -> dict[str, Any]: 

125 """Handle completion request.""" 

126 if not self.settings or not self.settings.enable_completions: 

127 return {"items": []} 

128 

129 uri = params["textDocument"]["uri"] 

130 position = params["position"] 

131 content = self._documents.get(uri, "") 

132 

133 if not self.syntax_support: 

134 return {"items": []} 

135 

136 # Get completions 

137 completions = self.syntax_support.get_completions( 

138 content, position["line"], position["character"] 

139 ) 

140 

141 items = [] 

142 for completion in completions: 

143 item = { 

144 "label": completion.label, 

145 "kind": self._completion_kind_to_lsp(completion.kind), 

146 "detail": completion.detail, 

147 "documentation": completion.documentation, 

148 "insertText": completion.insert_text or completion.label, 

149 "sortText": f"{100 - completion.priority:03d}_{completion.label}", 

150 } 

151 

152 # Add snippet support 

153 if completion.insert_text and "$" in completion.insert_text: 

154 item["insertTextFormat"] = 2 # Snippet 

155 

156 items.append(item) 

157 

158 return {"items": items} 

159 

160 async def text_document_hover( 

161 self, params: dict[str, Any] 

162 ) -> dict[str, Any] | None: 

163 """Handle hover request.""" 

164 if not self.settings or not self.settings.enable_hover: 

165 return None 

166 

167 uri = params["textDocument"]["uri"] 

168 position = params["position"] 

169 content = self._documents.get(uri, "") 

170 

171 if not self.syntax_support: 

172 return None 

173 

174 hover_info = self.syntax_support.get_hover_info( 

175 content, position["line"], position["character"] 

176 ) 

177 

178 if hover_info: 

179 return {"contents": {"kind": "markdown", "value": hover_info}} 

180 

181 return None 

182 

183 async def text_document_formatting( 

184 self, params: dict[str, Any] 

185 ) -> list[dict[str, Any]]: 

186 """Handle formatting request.""" 

187 if not self.settings or not self.settings.enable_formatting: 

188 return [] 

189 

190 uri = params["textDocument"]["uri"] 

191 content = self._documents.get(uri, "") 

192 

193 if not self.syntax_support: 

194 return [] 

195 

196 formatted = self.syntax_support.format_template(content) 

197 

198 if formatted != content: 

199 lines = content.split("\n") 

200 return [ 

201 { 

202 "range": { 

203 "start": {"line": 0, "character": 0}, 

204 "end": {"line": len(lines), "character": 0}, 

205 }, 

206 "newText": formatted, 

207 } 

208 ] 

209 

210 return [] 

211 

212 async def _run_diagnostics(self, uri: str, content: str) -> None: 

213 """Run diagnostics on document.""" 

214 if not self.syntax_support: 

215 return 

216 

217 errors = self.syntax_support.check_syntax(content) 

218 diagnostics = [] 

219 

220 for error in errors: 

221 diagnostic = { 

222 "range": { 

223 "start": {"line": error.line, "character": error.column}, 

224 "end": {"line": error.line, "character": error.column + 10}, 

225 }, 

226 "severity": self._severity_to_lsp(error.severity), 

227 "code": error.code, 

228 "source": "FastBlocks", 

229 "message": error.message, 

230 } 

231 

232 if error.fix_suggestion: 

233 t.cast(dict[str, t.Any], diagnostic)["data"] = { 

234 "fix": error.fix_suggestion 

235 } 

236 

237 diagnostics.append(diagnostic) 

238 

239 self._diagnostics[uri] = diagnostics 

240 

241 # In a real LSP implementation, you would send this to the client 

242 # self.send_notification("textDocument/publishDiagnostics", { 

243 # "uri": uri, 

244 # "diagnostics": diagnostics 

245 # }) 

246 

247 def _completion_kind_to_lsp(self, kind: str) -> int: 

248 """Convert completion kind to LSP constants.""" 

249 mapping = { 

250 "function": 3, # Function 

251 "variable": 6, # Variable 

252 "filter": 12, # Value 

253 "block": 14, # Keyword 

254 "component": 9, # Module 

255 "snippet": 15, # Snippet 

256 } 

257 return mapping.get(kind, 1) # Text 

258 

259 def _severity_to_lsp(self, severity: str) -> int: 

260 """Convert severity to LSP constants.""" 

261 mapping = {"error": 1, "warning": 2, "info": 3, "hint": 4} 

262 return mapping.get(severity, 1) 

263 

264 def get_current_diagnostics(self, uri: str) -> list[dict[str, Any]]: 

265 """Get current diagnostics for a document.""" 

266 return self._diagnostics.get(uri, []) 

267 

268 async def shutdown(self) -> None: 

269 """Handle server shutdown.""" 

270 self._documents.clear() 

271 self._diagnostics.clear() 

272 

273 

274class FastBlocksLanguageClient: 

275 """Simple language client for testing and integration.""" 

276 

277 def __init__(self) -> None: 

278 """Initialize language client.""" 

279 self.server = FastBlocksLanguageServer() 

280 self._initialized = False 

281 

282 async def initialize(self) -> None: 

283 """Initialize the language server.""" 

284 if self._initialized: 

285 return 

286 

287 await self.server.initialize( 

288 { 

289 "processId": None, 

290 "clientInfo": {"name": "FastBlocks Client", "version": "1.0.0"}, 

291 "capabilities": {}, 

292 } 

293 ) 

294 self._initialized = True 

295 

296 async def open_document(self, uri: str, content: str) -> None: 

297 """Open a document.""" 

298 await self.initialize() 

299 await self.server.text_document_did_open( 

300 { 

301 "textDocument": { 

302 "uri": uri, 

303 "languageId": "fastblocks", 

304 "version": 1, 

305 "text": content, 

306 } 

307 } 

308 ) 

309 

310 async def change_document(self, uri: str, content: str) -> None: 

311 """Change document content.""" 

312 await self.server.text_document_did_change( 

313 { 

314 "textDocument": {"uri": uri, "version": 2}, 

315 "contentChanges": [{"text": content}], 

316 } 

317 ) 

318 

319 async def get_completions( 

320 self, uri: str, line: int, character: int 

321 ) -> list[dict[str, Any]]: 

322 """Get completions at position.""" 

323 result = await self.server.text_document_completion( 

324 { 

325 "textDocument": {"uri": uri}, 

326 "position": {"line": line, "character": character}, 

327 } 

328 ) 

329 return t.cast(list[dict[str, Any]], result.get("items", [])) 

330 

331 async def get_hover( 

332 self, uri: str, line: int, character: int 

333 ) -> dict[str, Any] | None: 

334 """Get hover information.""" 

335 return await self.server.text_document_hover( 

336 { 

337 "textDocument": {"uri": uri}, 

338 "position": {"line": line, "character": character}, 

339 } 

340 ) 

341 

342 async def format_document(self, uri: str) -> list[dict[str, Any]]: 

343 """Format document.""" 

344 return await self.server.text_document_formatting( 

345 {"textDocument": {"uri": uri}} 

346 ) 

347 

348 def get_diagnostics(self, uri: str) -> list[dict[str, Any]]: 

349 """Get current diagnostics.""" 

350 return self.server.get_current_diagnostics(uri) 

351 

352 

353# VS Code extension configuration generator 

354def generate_vscode_extension() -> dict[str, Any]: 

355 """Generate VS Code extension configuration for FastBlocks.""" 

356 return { 

357 "name": "fastblocks-language-support", 

358 "displayName": "FastBlocks Language Support", 

359 "description": "Language support for FastBlocks templates", 

360 "version": "1.0.0", 

361 "publisher": "fastblocks", 

362 "engines": {"vscode": "^1.74.0"}, 

363 "categories": ["Programming Languages"], 

364 "activationEvents": ["onLanguage:fastblocks"], 

365 "main": "./out/extension.js", 

366 "contributes": { 

367 "languages": [ 

368 { 

369 "id": "fastblocks", 

370 "aliases": ["FastBlocks", "fastblocks"], 

371 "extensions": [".fb.html", ".fastblocks"], 

372 "configuration": "./language-configuration.json", 

373 } 

374 ], 

375 "grammars": [ 

376 { 

377 "language": "fastblocks", 

378 "scopeName": "text.html.fastblocks", 

379 "path": "./syntaxes/fastblocks.tmLanguage.json", 

380 } 

381 ], 

382 "configuration": { 

383 "type": "object", 

384 "title": "FastBlocks", 

385 "properties": { 

386 "fastblocks.languageServer.enabled": { 

387 "type": "boolean", 

388 "default": True, 

389 "description": "Enable FastBlocks language server", 

390 }, 

391 "fastblocks.languageServer.port": { 

392 "type": "number", 

393 "default": 7777, 

394 "description": "Language server port", 

395 }, 

396 "fastblocks.completion.enabled": { 

397 "type": "boolean", 

398 "default": True, 

399 "description": "Enable auto-completion", 

400 }, 

401 "fastblocks.diagnostics.enabled": { 

402 "type": "boolean", 

403 "default": True, 

404 "description": "Enable error checking", 

405 }, 

406 }, 

407 }, 

408 }, 

409 "scripts": { 

410 "vscode:prepublish": "npm run compile", 

411 "compile": "tsc -p ./", 

412 "watch": "tsc -watch -p ./", 

413 }, 

414 "devDependencies": { 

415 "@types/vscode": "^1.74.0", 

416 "@typescript-eslint/eslint-plugin": "^5.45.0", 

417 "@typescript-eslint/parser": "^5.45.0", 

418 "eslint": "^8.28.0", 

419 "typescript": "^4.9.4", 

420 }, 

421 "dependencies": {"vscode-languageclient": "^8.1.0"}, 

422 } 

423 

424 

425# TextMate grammar for syntax highlighting 

426def generate_textmate_grammar() -> dict[str, Any]: 

427 """Generate TextMate grammar for FastBlocks syntax highlighting.""" 

428 return { 

429 "$schema": "https://raw.githubusercontent.com/martinring/tmlanguage/master/tmlanguage.json", 

430 "name": "FastBlocks", 

431 "scopeName": "text.html.fastblocks", 

432 "patterns": [ 

433 {"include": "#fastblocks-variable"}, 

434 {"include": "#fastblocks-block"}, 

435 {"include": "#fastblocks-comment"}, 

436 {"include": "text.html.basic"}, 

437 ], 

438 "repository": { 

439 "fastblocks-variable": { 

440 "name": "meta.tag.template.value.fastblocks", 

441 "begin": r"\[\[", 

442 "end": r"\]\]", 

443 "beginCaptures": { 

444 "0": {"name": "punctuation.definition.tag.begin.fastblocks"} 

445 }, 

446 "endCaptures": { 

447 "0": {"name": "punctuation.definition.tag.end.fastblocks"} 

448 }, 

449 "patterns": [ 

450 {"include": "#fastblocks-filter"}, 

451 {"include": "#fastblocks-string"}, 

452 {"include": "#fastblocks-identifier"}, 

453 ], 

454 }, 

455 "fastblocks-block": { 

456 "name": "meta.tag.template.block.fastblocks", 

457 "begin": r"\[%", 

458 "end": r"%\]", 

459 "beginCaptures": { 

460 "0": {"name": "punctuation.definition.tag.begin.fastblocks"} 

461 }, 

462 "endCaptures": { 

463 "0": {"name": "punctuation.definition.tag.end.fastblocks"} 

464 }, 

465 "patterns": [ 

466 { 

467 "name": "keyword.control.fastblocks", 

468 "match": r"\b(if|else|elif|endif|for|endfor|block|endblock|extends|include|set|macro|endmacro)\b", 

469 }, 

470 {"include": "#fastblocks-string"}, 

471 {"include": "#fastblocks-identifier"}, 

472 ], 

473 }, 

474 "fastblocks-comment": { 

475 "name": "comment.block.fastblocks", 

476 "begin": r"\[#", 

477 "end": r"#\]", 

478 "beginCaptures": { 

479 "0": {"name": "punctuation.definition.comment.begin.fastblocks"} 

480 }, 

481 "endCaptures": { 

482 "0": {"name": "punctuation.definition.comment.end.fastblocks"} 

483 }, 

484 }, 

485 "fastblocks-filter": { 

486 "name": "support.function.filter.fastblocks", 

487 "match": r"\|\s*(\w+)", 

488 "captures": {"1": {"name": "entity.name.function.filter.fastblocks"}}, 

489 }, 

490 "fastblocks-string": { 

491 "name": "string.quoted.double.fastblocks", 

492 "begin": r'"', 

493 "end": r'"', 

494 "patterns": [ 

495 {"name": "constant.character.escape.fastblocks", "match": r"\\."} 

496 ], 

497 }, 

498 "fastblocks-identifier": { 

499 "name": "variable.other.fastblocks", 

500 "match": r"\b[a-zA-Z_][a-zA-Z0-9_]*\b", 

501 }, 

502 }, 

503 } 

504 

505 

506# ACB 0.19.0+ compatibility 

507__all__ = [ 

508 "FastBlocksLanguageServer", 

509 "FastBlocksLanguageClient", 

510 "LanguageServerSettings", 

511 "generate_vscode_extension", 

512 "generate_textmate_grammar", 

513]