refactor: update script execution with sys.executable

Updated the script execution commands to use `sys.executable` instead
of a hard-coded "python" to ensure compatibility with different Python
environments. Additionally, refactored code for better readability
and consistency in formatting, particularly around argument parsing
and error handling in both manage_hetzner_servers.py and
update_local_config.py.

Improves deployment flexibility and code maintainability.
This commit is contained in:
Kumi 2024-07-18 19:07:02 +02:00
parent 776e3e00ea
commit 5a64f31a1c
Signed by: kumi
GPG key ID: ECBCC9082395383F
2 changed files with 72 additions and 26 deletions

View file

@ -3,29 +3,40 @@ import configparser
from hcloud import Client from hcloud import Client
import argparse import argparse
import pathlib import pathlib
import os
import sys
UPDATE_SCRIPT_PATH = pathlib.Path(__file__).parent / "update_local_config.py" UPDATE_SCRIPT_PATH = pathlib.Path(__file__).parent / "update_local_config.py"
CONFIG_PATH = pathlib.Path(__file__).parent / "config.ini" CONFIG_PATH = pathlib.Path(__file__).parent / "config.ini"
def get_hetzner_client(): def get_hetzner_client():
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read(CONFIG_PATH) config.read(CONFIG_PATH)
api_token = config["hetzner"]["api_token"] api_token = config["hetzner"]["api_token"]
return Client(token=api_token) return Client(token=api_token)
def list_hetzner_servers(client): def list_hetzner_servers(client):
return client.servers.get_all() return client.servers.get_all()
def delete_hetzner_server(client, server): def delete_hetzner_server(client, server):
server.delete() server.delete()
def run_update_script(provider, location, server_type, interface, config_path): def run_update_script(provider, location, server_type, interface, config_path):
command = [ command = [
"python", UPDATE_SCRIPT_PATH, os.path.realpath(sys.executable),
"--provider", provider, UPDATE_SCRIPT_PATH,
"--location", location, "--provider",
"--server_type", server_type, provider,
"--interface", interface "--location",
location,
"--server_type",
server_type,
"--interface",
interface,
] ]
if config_path: if config_path:
@ -36,12 +47,19 @@ def run_update_script(provider, location, server_type, interface, config_path):
raise Exception(f"Error executing update script: {result.stderr}") raise Exception(f"Error executing update script: {result.stderr}")
print(result.stdout) print(result.stdout)
def main(): def main():
parser = argparse.ArgumentParser(description="Manage Hetzner servers and update local Wireguard VPN.") parser = argparse.ArgumentParser(
description="Manage Hetzner servers and update local Wireguard VPN."
)
parser.add_argument("--location", type=str, required=True, help="Server location") parser.add_argument("--location", type=str, required=True, help="Server location")
parser.add_argument("--server_type", type=str, required=True, help="Server type") parser.add_argument("--server_type", type=str, required=True, help="Server type")
parser.add_argument("--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)") parser.add_argument(
parser.add_argument("--config_path", type=str, help="Path to the Wireguard configuration file") "--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)"
)
parser.add_argument(
"--config_path", type=str, help="Path to the Wireguard configuration file"
)
args = parser.parse_args() args = parser.parse_args()
@ -63,5 +81,6 @@ def main():
delete_hetzner_server(client, server) delete_hetzner_server(client, server)
print(f"Deleted server: {server.name}") print(f"Deleted server: {server.name}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View file

@ -2,46 +2,70 @@ import subprocess
import re import re
import tempfile import tempfile
import os import os
import sys
import argparse import argparse
import pathlib import pathlib
SCRIPT_PATH = pathlib.Path(__file__).parent / "worker.py" SCRIPT_PATH = pathlib.Path(__file__).parent / "worker.py"
def execute_script(provider, location, server_type): def execute_script(provider, location, server_type):
command = [ command = [
"python", SCRIPT_PATH, os.path.realpath(sys.executable),
"--provider", provider, SCRIPT_PATH,
"--location", location, "--provider",
"--server_type", server_type provider,
"--location",
location,
"--server_type",
server_type,
] ]
result = subprocess.run(command, capture_output=True, text=True) result = subprocess.run(command, capture_output=True, text=True)
if result.returncode != 0: if result.returncode != 0:
raise Exception(f"Error executing script: {result.stderr}") raise Exception(f"Error executing script: {result.stderr}")
return result.stdout.strip() return result.stdout.strip()
def replace_peer_section(config_content, new_peer_section): def replace_peer_section(config_content, new_peer_section):
# Regular expression to match the [Peer] section # Regular expression to match the [Peer] section
peer_section_pattern = re.compile(r'\[Peer\](?:\n(?!\[)[^\n]*)*') peer_section_pattern = re.compile(r"\[Peer\](?:\n(?!\[)[^\n]*)*")
# Find the existing [Peer] section # Find the existing [Peer] section
match = peer_section_pattern.search(config_content) match = peer_section_pattern.search(config_content)
if match: if match:
# Replace the existing [Peer] section with the new one # Replace the existing [Peer] section with the new one
updated_content = config_content[:match.start()] + new_peer_section + config_content[match.end():] updated_content = (
config_content[: match.start()]
+ new_peer_section
+ config_content[match.end() :]
)
else: else:
# If no [Peer] section exists, append the new one # If no [Peer] section exists, append the new one
updated_content = config_content + "\n" + new_peer_section updated_content = config_content + "\n" + new_peer_section
return updated_content return updated_content
def main(): def main():
parser = argparse.ArgumentParser(description="Update local Wireguard configuration with a new peer.") parser = argparse.ArgumentParser(
parser.add_argument("--provider", type=str, choices=["hetzner", "aws", "digitalocean", "azure"], required=True, help="Cloud provider") description="Update local Wireguard configuration with a new peer."
)
parser.add_argument(
"--provider",
type=str,
choices=["hetzner", "aws", "digitalocean", "azure"],
required=True,
help="Cloud provider",
)
parser.add_argument("--location", type=str, required=True, help="Server location") parser.add_argument("--location", type=str, required=True, help="Server location")
parser.add_argument("--server_type", type=str, required=True, help="Server type") parser.add_argument("--server_type", type=str, required=True, help="Server type")
parser.add_argument("--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)") parser.add_argument(
parser.add_argument("--config_path", type=str, help="Path to the Wireguard configuration file") "--interface", type=str, required=True, help="Wireguard interface (e.g., wg0)"
)
parser.add_argument(
"--config_path", type=str, help="Path to the Wireguard configuration file"
)
args = parser.parse_args() args = parser.parse_args()
@ -55,7 +79,7 @@ def main():
new_peer_section = execute_script(provider, location, server_type) new_peer_section = execute_script(provider, location, server_type)
# Step 2: Read the local Wireguard configuration file # Step 2: Read the local Wireguard configuration file
with open(config_path, 'r') as file: with open(config_path, "r") as file:
config_content = file.read() config_content = file.read()
# Step 3: Replace the existing [Peer] section with the new one # Step 3: Replace the existing [Peer] section with the new one
@ -66,7 +90,7 @@ def main():
temp_file = pathlib.Path(tempdir) / f"{interface}.conf" temp_file = pathlib.Path(tempdir) / f"{interface}.conf"
temp_file_path = str(temp_file) temp_file_path = str(temp_file)
with open(temp_file_path, 'w') as file: with open(temp_file_path, "w") as file:
file.write(updated_content) file.write(updated_content)
# Step 5: Apply the updated Wireguard configuration # Step 5: Apply the updated Wireguard configuration
@ -76,7 +100,10 @@ def main():
# Overwrite the original config file with the updated content # Overwrite the original config file with the updated content
os.replace(temp_file_path, config_path) os.replace(temp_file_path, config_path)
print(f"Local Wireguard configuration for {interface} updated and applied successfully.") print(
f"Local Wireguard configuration for {interface} updated and applied successfully."
)
if __name__ == "__main__": if __name__ == "__main__":
main() main()