path_ops.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249
  1. """Path operations."""
  2. from __future__ import annotations
  3. import json
  4. import os
  5. import re
  6. import shutil
  7. from pathlib import Path
  8. from reflex import constants
  9. # Shorthand for join.
  10. join = os.linesep.join
  11. def rm(path: str | Path):
  12. """Remove a file or directory.
  13. Args:
  14. path: The path to the file or directory.
  15. """
  16. path = Path(path)
  17. if path.is_dir():
  18. shutil.rmtree(path)
  19. elif path.is_file():
  20. path.unlink()
  21. def cp(src: str | Path, dest: str | Path, overwrite: bool = True) -> bool:
  22. """Copy a file or directory.
  23. Args:
  24. src: The path to the file or directory.
  25. dest: The path to the destination.
  26. overwrite: Whether to overwrite the destination.
  27. Returns:
  28. Whether the copy was successful.
  29. """
  30. src, dest = Path(src), Path(dest)
  31. if src == dest:
  32. return False
  33. if not overwrite and dest.exists():
  34. return False
  35. if src.is_dir():
  36. rm(dest)
  37. shutil.copytree(src, dest)
  38. else:
  39. shutil.copyfile(src, dest)
  40. return True
  41. def mv(src: str | Path, dest: str | Path, overwrite: bool = True) -> bool:
  42. """Move a file or directory.
  43. Args:
  44. src: The path to the file or directory.
  45. dest: The path to the destination.
  46. overwrite: Whether to overwrite the destination.
  47. Returns:
  48. Whether the move was successful.
  49. """
  50. src, dest = Path(src), Path(dest)
  51. if src == dest:
  52. return False
  53. if not overwrite and dest.exists():
  54. return False
  55. rm(dest)
  56. shutil.move(src, dest)
  57. return True
  58. def mkdir(path: str | Path):
  59. """Create a directory.
  60. Args:
  61. path: The path to the directory.
  62. """
  63. Path(path).mkdir(parents=True, exist_ok=True)
  64. def ls(path: str | Path) -> list[Path]:
  65. """List the contents of a directory.
  66. Args:
  67. path: The path to the directory.
  68. Returns:
  69. A list of paths to the contents of the directory.
  70. """
  71. return list(Path(path).iterdir())
  72. def ln(src: str | Path, dest: str | Path, overwrite: bool = False) -> bool:
  73. """Create a symbolic link.
  74. Args:
  75. src: The path to the file or directory.
  76. dest: The path to the destination.
  77. overwrite: Whether to overwrite the destination.
  78. Returns:
  79. Whether the link was successful.
  80. """
  81. src, dest = Path(src), Path(dest)
  82. if src == dest:
  83. return False
  84. if not overwrite and (dest.exists() or dest.is_symlink()):
  85. return False
  86. if src.is_dir():
  87. rm(dest)
  88. src.symlink_to(dest, target_is_directory=True)
  89. else:
  90. src.symlink_to(dest)
  91. return True
  92. def which(program: str | Path) -> str | Path | None:
  93. """Find the path to an executable.
  94. Args:
  95. program: The name of the executable.
  96. Returns:
  97. The path to the executable.
  98. """
  99. return shutil.which(str(program))
  100. def use_system_install(var_name: str) -> bool:
  101. """Check if the system install should be used.
  102. Args:
  103. var_name: The name of the environment variable.
  104. Raises:
  105. ValueError: If the variable name is invalid.
  106. Returns:
  107. Whether the associated env var should use the system install.
  108. """
  109. if not var_name.startswith("REFLEX_USE_SYSTEM_"):
  110. raise ValueError("Invalid system install variable name.")
  111. return os.getenv(var_name, "").lower() in ["true", "1", "yes"]
  112. def use_system_node() -> bool:
  113. """Check if the system node should be used.
  114. Returns:
  115. Whether the system node should be used.
  116. """
  117. return use_system_install(constants.Node.USE_SYSTEM_VAR)
  118. def use_system_bun() -> bool:
  119. """Check if the system bun should be used.
  120. Returns:
  121. Whether the system bun should be used.
  122. """
  123. return use_system_install(constants.Bun.USE_SYSTEM_VAR)
  124. def get_node_bin_path() -> Path | None:
  125. """Get the node binary dir path.
  126. Returns:
  127. The path to the node bin folder.
  128. """
  129. bin_path = Path(constants.Node.BIN_PATH)
  130. if not bin_path.exists():
  131. str_path = which("node")
  132. return Path(str_path).parent.resolve() if str_path else None
  133. return bin_path.resolve()
  134. def get_node_path() -> str | None:
  135. """Get the node binary path.
  136. Returns:
  137. The path to the node binary file.
  138. """
  139. node_path = Path(constants.Node.PATH)
  140. if use_system_node() or not node_path.exists():
  141. system_node_path = which("node")
  142. return str(system_node_path) if system_node_path else None
  143. return str(node_path)
  144. def get_npm_path() -> str | None:
  145. """Get npm binary path.
  146. Returns:
  147. The path to the npm binary file.
  148. """
  149. npm_path = Path(constants.Node.NPM_PATH)
  150. if use_system_node() or not npm_path.exists():
  151. system_npm_path = which("npm")
  152. return str(system_npm_path) if system_npm_path else None
  153. return str(npm_path)
  154. def update_json_file(file_path: str | Path, update_dict: dict[str, int | str]):
  155. """Update the contents of a json file.
  156. Args:
  157. file_path: the path to the JSON file.
  158. update_dict: object to update json.
  159. """
  160. fp = Path(file_path)
  161. # Create the file if it doesn't exist.
  162. fp.touch(exist_ok=True)
  163. # Create an empty json object if file is empty
  164. fp.write_text("{}") if fp.stat().st_size == 0 else None
  165. # Read the existing json object from the file.
  166. json_object = {}
  167. if fp.stat().st_size:
  168. with open(fp) as f:
  169. json_object = json.load(f)
  170. # Update the json object with the new data.
  171. json_object.update(update_dict)
  172. # Write the updated json object to the file
  173. with open(fp, "w") as f:
  174. json.dump(json_object, f, ensure_ascii=False)
  175. def find_replace(directory: str | Path, find: str, replace: str):
  176. """Recursively find and replace text in files in a directory.
  177. Args:
  178. directory: The directory to search.
  179. find: The text to find.
  180. replace: The text to replace.
  181. """
  182. directory = Path(directory)
  183. for root, _dirs, files in os.walk(directory):
  184. for file in files:
  185. filepath = Path(root, file)
  186. text = filepath.read_text(encoding="utf-8")
  187. text = re.sub(find, replace, text)
  188. filepath.write_text(text, encoding="utf-8")