پلتفرم Flask

(Flask Platform)

اتصال به ذخیره‌سازی ابری

بدون شک اتصال برنامه به یک ذخیره‌سازی ابری مطمئن برای نگهداری و ارائه فایل‌های استاتیک وب‌سایت یا داده‌های آپلود شده توسط کاربران، باعث اطمینان خاطر صاحبان کسب و کار و بهبود عملکرد برنامه‌ می‌شود.

فهرست عناوین:

• نصب AWS SDK
• دریافت کلیدها
• تنظیم متغیر های محیطی
• نحوه‌ی استفاده

در صورتی که تمایلی به خواندن آموزش متنی ندارید می‌توانید ویدیوی آموزشی زیر ‌را مشاهده کنید.

نصب AWS SDK

از آنجا که ذخیره‌سازی ابری لیارا یک سرویس Object storage سازگار با پروتکل S3 است، شما می‌توانید با استفاده از AWS SDK‌، در زبان‌ها و فریم‌ورک‌های مختلفی این فضای ذخیره‌سازی را مدیریت کنید. حال برای اتصال به ذخیره‌سازی ابری در Flask باید با اجرای دستور زیر، پکیج boto3 را نصب کنید.

pip install boto3

دریافت کلیدها

اگر باکت شما خصوصی باشد، برای دسترسی به باکت، نیاز به کلید دارید. برای ساخت کلید، به صفحه ذخیره‌سازی ابری بروید و طبق عکس‌ها کلیدخود را بسازید.

به قسمت کلیدها رفته:

یک کلید جدید بسازید.

کلیدهای ساخته‌شده را کپی کنید. توجه داشته باشید که SECRET_KEY تنها یک بار نمایش داده می‌شود و پس از آن باید کلید را درجایی مطمئن ذخیره کنید.

تنظیم متغیر‌های محیطی

در این مرحله باید کلیدها، نام باکت و endpoint لیارا را در فایل .env ذخیره کنید

LIARA_ENDPOINT=<Liara Bucket Endpoint>
LIARA_BUCKET_NAME=<Bucket Name>
LIARA_ACCESS_KEY=<Access Key>
LIARA_SECRET_KEY=<Secret Key>

لطفا توجه داشته‌باشید که باید هریک از مقادیر بالا را با متغیر های باکت‌خودتان‌ جایگزین کنید. به عنوان مثال، یک فایل .env می‌تواند به شکل زیر باشد:

LIARA_ENDPOINT="https://storage.iran.liara.space"
LIARA_BUCKET_NAME="my-personal-files
LIARA_ACCESS_KEY="nad4u71et9dgc3go"
LIARA_SECRET_KEY="82c963df-1122-4c31-868b-0124a28ad57d""

نحوه‌ی استفاده

راه‌اندازی اولیه و اتصال به باکت:

from dotenv import load_dotenv
import boto3
from botocore.exceptions import NoCredentialsError
from urllib.parse import quote

load_dotenv()

LIARA_ENDPOINT = os.getenv("LIARA_ENDPOINT")
LIARA_ACCESS_KEY = os.getenv("LIARA_ACCESS_KEY")
LIARA_SECRET_KEY = os.getenv("LIARA_SECRET_KEY")
LIARA_BUCKET_NAME = os.getenv("LIARA_BUCKET_NAME")

s3 = boto3.client(
    "s3",
    endpoint_url=LIARA_ENDPOINT,
    aws_access_key_id=LIARA_ACCESS_KEY,
    aws_secret_access_key=LIARA_SECRET_KEY,
)

آپلود فایل:

s3.upload_fileobj(file, LIARA_BUCKET_NAME, file.filename)

دانلود فایل:

s3.download_file(LIARA_BUCKET_NAME, filename_in_bucket, filename_in_system)

دریافت لیست فایل‌های یک باکت:

files = s3.list_objects(Bucket=LIARA_BUCKET_NAME)
for file in files["Contents"]:
    print(file["Key"])

دریافت signed URL:

pre_signed_url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": LIARA_BUCKET_NAME, "Key": filename},
    ExpiresIn=12 * 60 * 60,  # 12 hours
)

دریافت لینک دائمی فایل. توجه کنید که دسترسی باکت باید public باشد:

filename_encoded = quote(filename)
permanent_url = f"https://{LIARA_BUCKET_NAME}.{LIARA_ENDPOINT.replace('https://', '')}/{filename_encoded}"

سورس‌کد ‌کامل جهت ارتباط با ذخیره‌سازی ابری در اپ flask

import os
import boto3
from dotenv import load_dotenv
from flask import Flask, request
from botocore.exceptions import NoCredentialsError
from urllib.parse import quote

load_dotenv()

LIARA_ENDPOINT = os.getenv("LIARA_ENDPOINT")
LIARA_ACCESS_KEY = os.getenv("LIARA_ACCESS_KEY")
LIARA_SECRET_KEY = os.getenv("LIARA_SECRET_KEY")
LIARA_BUCKET_NAME = os.getenv("LIARA_BUCKET_NAME")

s3 = boto3.client(
    "s3",
    endpoint_url=LIARA_ENDPOINT,
    aws_access_key_id=LIARA_ACCESS_KEY,
    aws_secret_access_key=LIARA_SECRET_KEY,
)

app = Flask(__name__)


@app.route("/")
def index():
    urls = {"urls": list()}
    for url in app.url_map.iter_rules():
        if '/static/' not in str(url):
            urls["urls"].append(str(url))
    return {"api": "running", **urls}


@app.route("/upload", methods=["POST"])
def upload_file():
    response = {"message": ""}
    file = request.files["file"]
    if file:
        try:
            s3.upload_fileobj(file, LIARA_BUCKET_NAME, file.filename)
            response["message"] = "File uploaded."
        except NoCredentialsError:
            response["message"] = "Liara credentials not found."
        except Exception as e:
            response["message"] = str(e)
        finally:
            return response
    else:
        response["message"] = "No file selected."


@app.route("/download/<filename>")
def download_file(filename):
    response = {"message": ""}
    try:
        s3.download_file(LIARA_BUCKET_NAME, filename, filename)
        response["message"] = "File downloaded."
    except NoCredentialsError:
        response["message"] = "Liara credentials not found."
    except Exception as e:
        response["message"] = str(e)
    finally:
        return response


@app.route("/list")
def list_files():
    response = {"message": list()}
    try:
        files = s3.list_objects(Bucket=LIARA_BUCKET_NAME)
        for file in files["Contents"]:
            response["message"].append(file["Key"])
    except NoCredentialsError:
        response["message"] = "Liara credentials not found."
    except Exception as e:
        response["message"] = str(e)
    finally:
        return response


@app.route("/presigned-url/<filename>")
def get_presigned_url(filename):
    response = {"message": ""}
    try:
        pre_signed_url = s3.generate_presigned_url(
            "get_object",
            Params={"Bucket": LIARA_BUCKET_NAME, "Key": filename},
            ExpiresIn=12 * 60 * 60,  # 12 hours
        )
        response["message"] = pre_signed_url
    except NoCredentialsError:
        response["message"] = "Liara credentials not found."
    except Exception as e:
        response["message"] = str(e)
    finally:
        return response


@app.route("/permanent-url/<filename>")
def get_permanent_url(filename):
    # bucket should be public
    response = {"message": ""}
    try:
        filename_encoded = quote(filename)
        permanent_url = f"https://{LIARA_BUCKET_NAME}.{LIARA_ENDPOINT.replace('https://', '')}/{filename_encoded}"
        response["message"] = permanent_url
    except NoCredentialsError:
        response["message"] = "Liara credentials not found."
    except Exception as e:
        response["message"] = str(e)
    finally:
        return response


if __name__ == "__main__":
    app.run()