dpzc_3_2/deploy.py

from hcloud import Client
from hcloud.networks.client import BoundNetwork
from hcloud.servers.client import BoundServer
from hcloud.ssh_keys.client import BoundSSHKey
from hcloud.networks.domain import NetworkSubnet
from hcloud.locations.domain import Location
from hcloud.images.domain import Image
from hcloud.server_types.domain import ServerType
from constants import *


def main():
    with open(API_KEY_FNAME) as f:
        api_key = f.read()

    client = Client(api_key)

    ssh_key = get_ssh_key(client)
    vnet = get_vnet(client)
    _ = create_db_server(client, vnet, ssh_key)
    _ = create_gitea_server(client, vnet, ssh_key)


def get_ssh_key(client: Client) -> BoundSSHKey:
    ssh_key = client.ssh_keys.get_by_name(SSH_KEY_NAME)
    if ssh_key is None:
        with open(SSH_KEY_FNAME) as f:
            ssh_key_pub = f.read()
        ssh_key = client.ssh_keys.create(
            name=SSH_KEY_NAME, public_key=ssh_key_pub)
    print(
        f"Klucz SSH: {ssh_key.data_model.name}, {ssh_key.data_model.public_key}")
    return ssh_key


def get_vnet(client: Client) -> BoundNetwork:
    vnet = client.networks.get_by_name(VNET_NAME)
    if vnet is None:
        vnet = client.networks.create(
            name=VNET_NAME,
            ip_range=IP_RANGE,
            subnets=[
                NetworkSubnet(ip_range=IP_RANGE,
                              network_zone=NETWORK_ZONE, type=NETWORK_TYPE)
            ]
        )
    print(
        f"Sieć wirtualna: {vnet.data_model.name} ({vnet.data_model.ip_range})")
    return vnet


def create_db_server(client: Client, vnet: BoundNetwork, ssh_key: BoundSSHKey) -> BoundServer:
    db_server = client.servers.get_by_name(DB_SERVER_NAME)
    if db_server is not None:
        print(
            f"Serwer: {db_server.data_model.name}\n\tpubliczne IP: {db_server.data_model.public_net.ipv4.ip}\n\tprywatne IP: {db_server.data_model.private_net[0].ip}")
        return db_server
    with open(CLOUD_INIT_DB_FNAME) as f:
        cloud_init_db = f.read()
    db_server_resp = client.servers.create(
        name=DB_SERVER_NAME,
        server_type=ServerType(MACHINE_TYPE),
        image=Image(name=MACHINE_OS),
        ssh_keys=[ssh_key],
        networks=[vnet],
        location=Location(MACHINE_LOCATION),
        user_data=cloud_init_db
    )

    db_server_resp.action.wait_until_finished()
    print(f"Tworzenie serwera db: {db_server_resp.action.complete}")

    db_server = client.servers.get_by_name(DB_SERVER_NAME)
    print(
        f"Serwer: {db_server.data_model.name}\n\tpubliczne IP: {db_server.data_model.public_net.ipv4.ip}\n\tprywatne IP: {db_server.data_model.private_net[0].ip}")

    return db_server


def create_gitea_server(client: Client, vnet: BoundNetwork, ssh_key: BoundSSHKey) -> BoundServer:
    gitea_server = client.servers.get_by_name(GITEA_SERVER_NAME)
    if gitea_server is not None:
        print(
            f"Serwer: {gitea_server.data_model.name}\n\tpubliczne IP: {gitea_server.data_model.public_net.ipv4.ip}\n\tprywatne IP: {gitea_server.data_model.private_net[0].ip}")
        return gitea_server
    with open(CLOUD_INIT_GITEA_FNAME) as f:
        cloud_init_gitea = f.read()
    gitea_server_resp = client.servers.create(
        name=GITEA_SERVER_NAME,
        server_type=ServerType(MACHINE_TYPE),
        image=Image(name=MACHINE_OS),
        ssh_keys=[ssh_key],
        networks=[vnet],
        location=Location(MACHINE_LOCATION),
        user_data=cloud_init_gitea
    )
    gitea_server_resp.action.wait_until_finished()
    print(f"Tworzenie serwera gitea: {gitea_server_resp.action.complete}")
    gitea_server = client.servers.get_by_name(GITEA_SERVER_NAME)
    print(
        f"Serwer: {gitea_server.data_model.name}\n\tpubliczne IP: {gitea_server.data_model.public_net.ipv4.ip}\n\tprywatne IP: {gitea_server.data_model.private_net[0].ip}")
    return gitea_server


if __name__ == '__main__':
    main()