Last active
July 6, 2021 16:19
-
-
Save t0mm4rx/3db93097d811db503d02bb0f78850917 to your computer and use it in GitHub Desktop.
Automatically deploy appwrite Python functions
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Deploy API functions to Appwrite. | |
This script will loop though your ./api folder, get all .py files, then: | |
- check if this function version has been deployed | |
- if not, will create or update the function | |
Functions name will be the name of the file, without ".py". | |
By default functions have no execution permission, you need to change | |
it manually. | |
You should have a .env file with following env variables: | |
- _APP_DOMAIN | |
- PROJECT_ID | |
- ADMIN_KEY (API key) | |
Also make sure /tmp is writable by this script, or change the temporary archive location. | |
""" | |
from appwrite.client import Client | |
from appwrite.services.functions import Functions | |
import dotenv | |
import os | |
import json | |
import hashlib | |
client = Client() | |
print(f"Connecting to appwrite: https://{dotenv.get_key('.env', '_APP_DOMAIN')}/v1") | |
client.set_endpoint(f"https://{dotenv.get_key('.env', '_APP_DOMAIN')}/v1") | |
client.set_project(dotenv.get_key(".env", "PROJECT_ID")) | |
client.set_key(dotenv.get_key(".env", "ADMIN_KEY")) | |
functions = Functions(client) | |
os.system("cp ./.env ./api/.env") | |
os.system("cd ./api && PIP_TARGET=./.appwrite pip install -r ./requirements.txt --upgrade --ignore-installed") | |
os.system("cd ./api && tar -zcf /tmp/functions.tar.gz ./") | |
def get_function(functions, name): | |
"""Return function ID if exists.""" | |
for f in functions: | |
if (f["name"] == name): | |
return f["$id"] | |
return None | |
def get_hashes(): | |
"""Get the latest hash published.""" | |
try: | |
with open("./.functions_hashes.json") as file: | |
return json.load(file) | |
except: | |
return {} | |
def save_hashes(hashes): | |
"""Save the latest hash published.""" | |
with open("./.functions_hashes.json", "w+") as file: | |
return json.dump(hashes, file) | |
existing_functions = functions.list()["functions"] | |
latest_hashes = get_hashes() | |
for file in os.listdir("./api"): | |
if (file.endswith(".py")): | |
name = file.removesuffix(".py") | |
print(f"🐍 Function {name}") | |
hash = hashlib.md5(open(f"./api/{file}", "rb").read()).hexdigest() | |
if (name in latest_hashes and latest_hashes[name] == hash): | |
print("ℹ️ No change") | |
continue | |
latest_hashes[name] = hash | |
print("ℹ️ Changes detected, will deploy...") | |
function_id = get_function(existing_functions, name) | |
if (function_id is None): | |
function_id = functions.create(name, [], "python-3.9")["$id"] | |
tag = functions.create_tag(function_id, f"python ./{file}", open("/tmp/functions.tar.gz", "rb")) | |
tag_id = tag["$id"] | |
functions.update_tag(function_id, tag_id) | |
print("✅ Deployed") | |
save_hashes(latest_hashes) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment