use crate::*;use anyhow::Context as _;use dap::{DebugRequest, StartDebuggingRequestArguments, adapters::DebugTaskDefinition};use fs::RemoveOptions;use futures::{StreamExt, TryStreamExt};use gpui::http_client::AsyncBody;use gpui::{AsyncApp, SharedString};use json_dotpath::DotPaths;use language::LanguageName;use paths::debug_adapters_dir;use serde_json::Value;use smol::fs::File;use smol::io::AsyncReadExt;use smol::lock::OnceCell;use std::ffi::OsString;use std::net::Ipv4Addr;use std::str::FromStr;use std::{ collections::HashMap, ffi::OsStr, path::{Path, PathBuf},};use util::{ResultExt, maybe}; #[derive(Default)]pub(crate) struct PythonDebugAdapter { base_venv_path: OnceCell<Result<Arc<Path>, String>>, debugpy_whl_base_path: OnceCell<Result<Arc<Path>, String>>,} impl PythonDebugAdapter { const ADAPTER_NAME: &'static str = "Debugpy"; const DEBUG_ADAPTER_NAME: DebugAdapterName = DebugAdapterName(SharedString::new_static(Self::ADAPTER_NAME)); const LANGUAGE_NAME: &'static str = "Python"; async fn generate_debugpy_arguments( host: &Ipv4Addr, port: u16, user_installed_path: Option<&Path>, user_args: Option<Vec<String>>, ) -> Result<Vec<String>> { let mut args = if let Some(user_installed_path) = user_installed_path { log::debug!( "Using user-installed debugpy adapter from: {}", user_installed_path.display() ); vec![user_installed_path.to_string_lossy().to_string()] } else { let adapter_path = paths::debug_adapters_dir().join(Self::DEBUG_ADAPTER_NAME.as_ref()); let path = adapter_path .join("debugpy") .join("adapter") .to_string_lossy() .into_owned(); log::debug!("Using pip debugpy adapter from: {path}"); vec![path] }; args.extend(if let Some(args) = user_args { args } else { vec![format!("--host={}", host), format!("--port={}", port)] }); Ok(args) } async fn request_args( &self, delegate: &Arc<dyn DapDelegate>, task_definition: &DebugTaskDefinition, ) -> Result<StartDebuggingRequestArguments> { let request = self.request_kind(&task_definition.config).await?; let mut configuration = task_definition.config.clone(); if let Ok(console) = configuration.dot_get_mut("console") { // Use built-in Zed terminal if user did not explicitly provide a setting for console. if console.is_null() { *console = Value::String("integratedTerminal".into()); } } if let Some(obj) = configuration.as_object_mut() { obj.entry("cwd") .or_insert(delegate.worktree_root_path().to_string_lossy().into()); } Ok(StartDebuggingRequestArguments { configuration, request, }) } async fn fetch_wheel(&self, delegate: &Arc<dyn DapDelegate>) -> Result<Arc<Path>, String> { let download_dir = debug_adapters_dir().join(Self::ADAPTER_NAME).join("wheels"); std::fs::create_dir_all(&download_dir).map_err(|e| e.to_string())?; let system_python = self.base_venv_path(delegate).await?; let installation_succeeded = util::command::new_smol_command(system_python.as_ref()) .args([ "-m", "pip", "download", "debugpy", "--only-binary=:all:", "-d", download_dir.to_string_lossy().as_ref(), ]) .output() .await .map_err(|e| format!("{e}"))? .status .success(); if !installation_succeeded { return Err("debugpy installation failed (could not fetch Debugpy's wheel)".into()); } let wheel_path = std::fs::read_dir(&download_dir) .map_err(|e| e.to_string())? .find_map(|entry| { entry.ok().filter(|e| { e.file_type().is_ok_and(|typ| typ.is_file()) && Path::new(&e.file_name()).extension() == Some("whl".as_ref()) }) }) .ok_or_else(|| String::from("Did not find a .whl in {download_dir}"))?; util::archive::extract_zip( &debug_adapters_dir().join(Self::ADAPTER_NAME), File::open(&wheel_path.path()) .await .map_err(|e| e.to_string())?, ) .await .map_err(|e| e.to_string())?; Ok(Arc::from(wheel_path.path())) } async fn maybe_fetch_new_wheel(&self, delegate: &Arc<dyn DapDelegate>) { let latest_release = delegate .http_client() .get( "https://pypi.org/pypi/debugpy/json", AsyncBody::empty(), false, ) .await .log_err(); maybe!(async move { let response = latest_release.filter(|response| response.status().is_success())?; let download_dir = debug_adapters_dir().join(Self::ADAPTER_NAME); std::fs::create_dir_all(&download_dir).ok()?; let mut output = String::new(); response .into_body() .read_to_string(&mut output) .await .ok()?; let as_json = serde_json::Value::from_str(&output).ok()?; let latest_version = as_json.get("info").and_then(|info| { info.get("version") .and_then(|version| version.as_str()) .map(ToOwned::to_owned) })?; let dist_info_dirname: OsString = format!("debugpy-{latest_version}.dist-info").into(); let is_up_to_date = delegate .fs() .read_dir(&debug_adapters_dir().join(Self::ADAPTER_NAME)) .await .ok()? .into_stream() .any(async |entry| { entry.is_ok_and(|e| e.file_name().is_some_and(|name| name == dist_info_dirname)) }) .await; if !is_up_to_date { delegate .fs() .remove_dir( &debug_adapters_dir().join(Self::ADAPTER_NAME), RemoveOptions { recursive: true, ignore_if_not_exists: true, }, ) .await .ok()?; self.fetch_wheel(delegate).await.ok()?; } Some(()) }) .await; } async fn fetch_debugpy_whl( &self, delegate: &Arc<dyn DapDelegate>, ) -> Result<Arc<Path>, String> { self.debugpy_whl_base_path .get_or_init(|| async move { self.maybe_fetch_new_wheel(delegate).await; Ok(Arc::from( debug_adapters_dir() .join(Self::ADAPTER_NAME) .join("debugpy") .join("adapter") .as_ref(), )) }) .await .clone() } async fn base_venv_path(&self, delegate: &Arc<dyn DapDelegate>) -> Result<Arc<Path>, String> { self.base_venv_path .get_or_init(|| async { let base_python = Self::system_python_name(delegate) .await .ok_or_else(|| String::from("Could not find a Python installation"))?; let did_succeed = util::command::new_smol_command(base_python) .args(["-m", "venv", "zed_base_venv"]) .current_dir( paths::debug_adapters_dir().join(Self::DEBUG_ADAPTER_NAME.as_ref()), ) .spawn() .map_err(|e| format!("{e:#?}"))? .status() .await .map_err(|e| format!("{e:#?}"))? .success(); if !did_succeed { return Err("Failed to create base virtual environment".into()); } const DIR: &str = if cfg!(target_os = "windows") { "Scripts" } else { "bin" }; Ok(Arc::from( paths::debug_adapters_dir() .join(Self::DEBUG_ADAPTER_NAME.as_ref()) .join("zed_base_venv") .join(DIR) .join("python3") .as_ref(), )) }) .await .clone() } async fn system_python_name(delegate: &Arc<dyn DapDelegate>) -> Option<String> { const BINARY_NAMES: [&str; 3] = ["python3", "python", "py"]; let mut name = None; for cmd in BINARY_NAMES { name = delegate .which(OsStr::new(cmd)) .await .map(|path| path.to_string_lossy().to_string()); if name.is_some() { break; } } name } async fn get_installed_binary( &self, delegate: &Arc<dyn DapDelegate>, config: &DebugTaskDefinition, user_installed_path: Option<PathBuf>, user_args: Option<Vec<String>>, python_from_toolchain: Option<String>, ) -> Result<DebugAdapterBinary> { let tcp_connection = config.tcp_connection.clone().unwrap_or_default(); let (host, port, timeout) = crate::configure_tcp_connection(tcp_connection).await?; let python_path = if let Some(toolchain) = python_from_toolchain { Some(toolchain) } else { Self::system_python_name(delegate).await }; let python_command = python_path.context("failed to find binary path for Python")?; log::debug!("Using Python executable: {}", python_command); let arguments = Self::generate_debugpy_arguments( &host, port, user_installed_path.as_deref(), user_args, ) .await?; log::debug!( "Starting debugpy adapter with command: {} {}", python_command, arguments.join(" ") ); Ok(DebugAdapterBinary { command: Some(python_command), arguments, connection: Some(adapters::TcpArguments { host, port, timeout, }), cwd: Some(delegate.worktree_root_path().to_path_buf()), envs: HashMap::default(), request_args: self.request_args(delegate, config).await?, }) }} #[async_trait(?Send)]impl DebugAdapter for PythonDebugAdapter { fn name(&self) -> DebugAdapterName { Self::DEBUG_ADAPTER_NAME } fn adapter_language_name(&self) -> Option<LanguageName> { Some(SharedString::new_static("Python").into()) } async fn config_from_zed_format(&self, zed_scenario: ZedDebugConfig) -> Result<DebugScenario> { let mut args = json!({ "request": match zed_scenario.request { DebugRequest::Launch(_) => "launch", DebugRequest::Attach(_) => "attach", }, "subProcess": true, "redirectOutput": true, }); let map = args.as_object_mut().unwrap(); match &zed_scenario.request { DebugRequest::Attach(attach) => { map.insert("processId".into(), attach.process_id.into()); } DebugRequest::Launch(launch) => { map.insert("program".into(), launch.program.clone().into()); map.insert("args".into(), launch.args.clone().into()); if !launch.env.is_empty() { map.insert("env".into(), launch.env_json()); } if let Some(stop_on_entry) = zed_scenario.stop_on_entry { map.insert("stopOnEntry".into(), stop_on_entry.into()); } if let Some(cwd) = launch.cwd.as_ref() { map.insert("cwd".into(), cwd.to_string_lossy().into_owned().into()); } } } Ok(DebugScenario { adapter: zed_scenario.adapter, label: zed_scenario.label, config: args, build: None, tcp_connection: None, }) } fn dap_schema(&self) -> serde_json::Value { json!({ "properties": { "request": { "type": "string", "enum": ["attach", "launch"], "description": "Debug adapter request type" }, "autoReload": { "default": {}, "description": "Configures automatic reload of code on edit.", "properties": { "enable": { "default": false, "description": "Automatically reload code on edit.", "type": "boolean" }, "exclude": { "default": [ "**/.git/**", "**/.metadata/**", "**/__pycache__/**", "**/node_modules/**", "**/site-packages/**" ], "description": "Glob patterns of paths to exclude from auto reload.", "items": { "type": "string" }, "type": "array" }, "include": { "default": [ "**/*.py", "**/*.pyw" ], "description": "Glob patterns of paths to include in auto reload.", "items": { "type": "string" }, "type": "array" } }, "type": "object" }, "debugAdapterPath": { "description": "Path (fully qualified) to the python debug adapter executable.", "type": "string" }, "django": { "default": false, "description": "Django debugging.", "type": "boolean" }, "jinja": { "default": null, "description": "Jinja template debugging (e.g. Flask).", "enum": [ false, null, true ] }, "justMyCode": { "default": true, "description": "If true, show and debug only user-written code. If false, show and debug all code, including library calls.", "type": "boolean" }, "logToFile": { "default": false, "description": "Enable logging of debugger events to a log file. This file can be found in the debugpy extension install folder.", "type": "boolean" }, "pathMappings": { "default": [], "items": { "label": "Path mapping", "properties": { "localRoot": { "default": "${ZED_WORKTREE_ROOT}", "label": "Local source root.", "type": "string" }, "remoteRoot": { "default": "", "label": "Remote source root.", "type": "string" } }, "required": [ "localRoot", "remoteRoot" ], "type": "object" }, "label": "Path mappings.", "type": "array" }, "redirectOutput": { "default": true, "description": "Redirect output.", "type": "boolean" }, "showReturnValue": { "default": true, "description": "Show return value of functions when stepping.", "type": "boolean" }, "subProcess": { "default": false, "description": "Whether to enable Sub Process debugging", "type": "boolean" }, "consoleName": { "default": "Python Debug Console", "description": "Display name of the debug console or terminal", "type": "string" }, "clientOS": { "default": null, "description": "OS that VS code is using.", "enum": [ "windows", null, "unix" ] }Lines 1–500 of 839. Load more
use crate::*; use anyhow::Context as _; use dap::{DebugRequest, StartDebuggingRequestArguments, adapters::DebugTaskDefinition}; use fs::RemoveOptions; use futures::{StreamExt, TryStreamExt}; use gpui::http_client::AsyncBody; use gpui::{AsyncApp, SharedString}; use json_dotpath::DotPaths; use language::LanguageName; use paths::debug_adapters_dir; use serde_json::Value; use smol::fs::File; use smol::io::AsyncReadExt; use smol::lock::OnceCell; use std::ffi::OsString; use std::net::Ipv4Addr; use std::str::FromStr; use std::{ collections::HashMap, ffi::OsStr, path::{Path, PathBuf}, }; use util::{ResultExt, maybe}; #[derive(Default)] pub(crate) struct PythonDebugAdapter { base_venv_path: OnceCell<Result<Arc<Path>, String>>, debugpy_whl_base_path: OnceCell<Result<Arc<Path>, String>>, } impl PythonDebugAdapter { const ADAPTER_NAME: &'static str = "Debugpy"; const DEBUG_ADAPTER_NAME: DebugAdapterName = DebugAdapterName(SharedString::new_static(Self::ADAPTER_NAME)); const LANGUAGE_NAME: &'static str = "Python"; async fn generate_debugpy_arguments( host: &Ipv4Addr, port: u16, user_installed_path: Option<&Path>, user_args: Option<Vec<String>>, ) -> Result<Vec<String>> { let mut args = if let Some(user_installed_path) = user_installed_path { log::debug!( "Using user-installed debugpy adapter from: {}", user_installed_path.display() ); vec![user_installed_path.to_string_lossy().to_string()] } else { let adapter_path = paths::debug_adapters_dir().join(Self::DEBUG_ADAPTER_NAME.as_ref()); let path = adapter_path .join("debugpy") .join("adapter") .to_string_lossy() .into_owned(); log::debug!("Using pip debugpy adapter from: {path}"); vec![path] }; args.extend(if let Some(args) = user_args { args } else { vec![format!("--host={}", host), format!("--port={}", port)] }); Ok(args) } async fn request_args( &self, delegate: &Arc<dyn DapDelegate>, task_definition: &DebugTaskDefinition, ) -> Result<StartDebuggingRequestArguments> { let request = self.request_kind(&task_definition.config).await?; let mut configuration = task_definition.config.clone(); if let Ok(console) = configuration.dot_get_mut("console") { // Use built-in Zed terminal if user did not explicitly provide a setting for console. if console.is_null() { *console = Value::String("integratedTerminal".into()); } } if let Some(obj) = configuration.as_object_mut() { obj.entry("cwd") .or_insert(delegate.worktree_root_path().to_string_lossy().into()); } Ok(StartDebuggingRequestArguments { configuration, request, }) } async fn fetch_wheel(&self, delegate: &Arc<dyn DapDelegate>) -> Result<Arc<Path>, String> { let download_dir = debug_adapters_dir().join(Self::ADAPTER_NAME).join("wheels"); std::fs::create_dir_all(&download_dir).map_err(|e| e.to_string())?; let system_python = self.base_venv_path(delegate).await?; let installation_succeeded = util::command::new_smol_command(system_python.as_ref()) .args([ "-m", "pip", "download", "debugpy", "--only-binary=:all:", "-d", download_dir.to_string_lossy().as_ref(), ]) .output() .await .map_err(|e| format!("{e}"))? .status .success(); if !installation_succeeded { return Err("debugpy installation failed (could not fetch Debugpy's wheel)".into()); } let wheel_path = std::fs::read_dir(&download_dir) .map_err(|e| e.to_string())? .find_map(|entry| { entry.ok().filter(|e| { e.file_type().is_ok_and(|typ| typ.is_file()) && Path::new(&e.file_name()).extension() == Some("whl".as_ref()) }) }) .ok_or_else(|| String::from("Did not find a .whl in {download_dir}"))?; util::archive::extract_zip( &debug_adapters_dir().join(Self::ADAPTER_NAME), File::open(&wheel_path.path()) .await .map_err(|e| e.to_string())?, ) .await .map_err(|e| e.to_string())?; Ok(Arc::from(wheel_path.path())) } async fn maybe_fetch_new_wheel(&self, delegate: &Arc<dyn DapDelegate>) { let latest_release = delegate .http_client() .get( "https://pypi.org/pypi/debugpy/json", AsyncBody::empty(), false, ) .await .log_err(); maybe!(async move { let response = latest_release.filter(|response| response.status().is_success())?; let download_dir = debug_adapters_dir().join(Self::ADAPTER_NAME); std::fs::create_dir_all(&download_dir).ok()?; let mut output = String::new(); response .into_body() .read_to_string(&mut output) .await .ok()?; let as_json = serde_json::Value::from_str(&output).ok()?; let latest_version = as_json.get("info").and_then(|info| { info.get("version") .and_then(|version| version.as_str()) .map(ToOwned::to_owned) })?; let dist_info_dirname: OsString = format!("debugpy-{latest_version}.dist-info").into(); let is_up_to_date = delegate .fs() .read_dir(&debug_adapters_dir().join(Self::ADAPTER_NAME)) .await .ok()? .into_stream() .any(async |entry| { entry.is_ok_and(|e| e.file_name().is_some_and(|name| name == dist_info_dirname)) }) .await; if !is_up_to_date { delegate .fs() .remove_dir( &debug_adapters_dir().join(Self::ADAPTER_NAME), RemoveOptions { recursive: true, ignore_if_not_exists: true, }, ) .await .ok()?; self.fetch_wheel(delegate).await.ok()?; } Some(()) }) .await; } async fn fetch_debugpy_whl( &self, delegate: &Arc<dyn DapDelegate>, ) -> Result<Arc<Path>, String> { self.debugpy_whl_base_path .get_or_init(|| async move { self.maybe_fetch_new_wheel(delegate).await; Ok(Arc::from( debug_adapters_dir() .join(Self::ADAPTER_NAME) .join("debugpy") .join("adapter") .as_ref(), )) }) .await .clone() } async fn base_venv_path(&self, delegate: &Arc<dyn DapDelegate>) -> Result<Arc<Path>, String> { self.base_venv_path .get_or_init(|| async { let base_python = Self::system_python_name(delegate) .await .ok_or_else(|| String::from("Could not find a Python installation"))?; let did_succeed = util::command::new_smol_command(base_python) .args(["-m", "venv", "zed_base_venv"]) .current_dir( paths::debug_adapters_dir().join(Self::DEBUG_ADAPTER_NAME.as_ref()), ) .spawn() .map_err(|e| format!("{e:#?}"))? .status() .await .map_err(|e| format!("{e:#?}"))? .success(); if !did_succeed { return Err("Failed to create base virtual environment".into()); } const DIR: &str = if cfg!(target_os = "windows") { "Scripts" } else { "bin" }; Ok(Arc::from( paths::debug_adapters_dir() .join(Self::DEBUG_ADAPTER_NAME.as_ref()) .join("zed_base_venv") .join(DIR) .join("python3") .as_ref(), )) }) .await .clone() } async fn system_python_name(delegate: &Arc<dyn DapDelegate>) -> Option<String> { const BINARY_NAMES: [&str; 3] = ["python3", "python", "py"]; let mut name = None; for cmd in BINARY_NAMES { name = delegate .which(OsStr::new(cmd)) .await .map(|path| path.to_string_lossy().to_string()); if name.is_some() { break; } } name } async fn get_installed_binary( &self, delegate: &Arc<dyn DapDelegate>, config: &DebugTaskDefinition, user_installed_path: Option<PathBuf>, user_args: Option<Vec<String>>, python_from_toolchain: Option<String>, ) -> Result<DebugAdapterBinary> { let tcp_connection = config.tcp_connection.clone().unwrap_or_default(); let (host, port, timeout) = crate::configure_tcp_connection(tcp_connection).await?; let python_path = if let Some(toolchain) = python_from_toolchain { Some(toolchain) } else { Self::system_python_name(delegate).await }; let python_command = python_path.context("failed to find binary path for Python")?; log::debug!("Using Python executable: {}", python_command); let arguments = Self::generate_debugpy_arguments( &host, port, user_installed_path.as_deref(), user_args, ) .await?; log::debug!( "Starting debugpy adapter with command: {} {}", python_command, arguments.join(" ") ); Ok(DebugAdapterBinary { command: Some(python_command), arguments, connection: Some(adapters::TcpArguments { host, port, timeout, }), cwd: Some(delegate.worktree_root_path().to_path_buf()), envs: HashMap::default(), request_args: self.request_args(delegate, config).await?, }) } } #[async_trait(?Send)] impl DebugAdapter for PythonDebugAdapter { fn name(&self) -> DebugAdapterName { Self::DEBUG_ADAPTER_NAME } fn adapter_language_name(&self) -> Option<LanguageName> { Some(SharedString::new_static("Python").into()) } async fn config_from_zed_format(&self, zed_scenario: ZedDebugConfig) -> Result<DebugScenario> { let mut args = json!({ "request": match zed_scenario.request { DebugRequest::Launch(_) => "launch", DebugRequest::Attach(_) => "attach", }, "subProcess": true, "redirectOutput": true, }); let map = args.as_object_mut().unwrap(); match &zed_scenario.request { DebugRequest::Attach(attach) => { map.insert("processId".into(), attach.process_id.into()); } DebugRequest::Launch(launch) => { map.insert("program".into(), launch.program.clone().into()); map.insert("args".into(), launch.args.clone().into()); if !launch.env.is_empty() { map.insert("env".into(), launch.env_json()); } if let Some(stop_on_entry) = zed_scenario.stop_on_entry { map.insert("stopOnEntry".into(), stop_on_entry.into()); } if let Some(cwd) = launch.cwd.as_ref() { map.insert("cwd".into(), cwd.to_string_lossy().into_owned().into()); } } } Ok(DebugScenario { adapter: zed_scenario.adapter, label: zed_scenario.label, config: args, build: None, tcp_connection: None, }) } fn dap_schema(&self) -> serde_json::Value { json!({ "properties": { "request": { "type": "string", "enum": ["attach", "launch"], "description": "Debug adapter request type" }, "autoReload": { "default": {}, "description": "Configures automatic reload of code on edit.", "properties": { "enable": { "default": false, "description": "Automatically reload code on edit.", "type": "boolean" }, "exclude": { "default": [ "**/.git/**", "**/.metadata/**", "**/__pycache__/**", "**/node_modules/**", "**/site-packages/**" ], "description": "Glob patterns of paths to exclude from auto reload.", "items": { "type": "string" }, "type": "array" }, "include": { "default": [ "**/*.py", "**/*.pyw" ], "description": "Glob patterns of paths to include in auto reload.", "items": { "type": "string" }, "type": "array" } }, "type": "object" }, "debugAdapterPath": { "description": "Path (fully qualified) to the python debug adapter executable.", "type": "string" }, "django": { "default": false, "description": "Django debugging.", "type": "boolean" }, "jinja": { "default": null, "description": "Jinja template debugging (e.g. Flask).", "enum": [ false, null, true ] }, "justMyCode": { "default": true, "description": "If true, show and debug only user-written code. If false, show and debug all code, including library calls.", "type": "boolean" }, "logToFile": { "default": false, "description": "Enable logging of debugger events to a log file. This file can be found in the debugpy extension install folder.", "type": "boolean" }, "pathMappings": { "default": [], "items": { "label": "Path mapping", "properties": { "localRoot": { "default": "${ZED_WORKTREE_ROOT}", "label": "Local source root.", "type": "string" }, "remoteRoot": { "default": "", "label": "Remote source root.", "type": "string" } }, "required": [ "localRoot", "remoteRoot" ], "type": "object" }, "label": "Path mappings.", "type": "array" }, "redirectOutput": { "default": true, "description": "Redirect output.", "type": "boolean" }, "showReturnValue": { "default": true, "description": "Show return value of functions when stepping.", "type": "boolean" }, "subProcess": { "default": false, "description": "Whether to enable Sub Process debugging", "type": "boolean" }, "consoleName": { "default": "Python Debug Console", "description": "Display name of the debug console or terminal", "type": "string" }, "clientOS": { "default": null, "description": "OS that VS code is using.", "enum": [ "windows", null, "unix" ] }