اتصال به فضای ذخیره‌سازی ابری در برنامه‌های Flask

pip install boto3

LIARA_ENDPOINT_URL=https://storage.c2.liara.site
LIARA_ACCESS_KEY=nieiou08cnbod58p
LIARA_SECRET_KEY=20b71a4c-1168-4945-8ed3-4724dbf9e997
BUCKET_NAME=my-bucket

from flask import Flask, render_template, request, flash
import boto3
from botocore.exceptions import NoCredentialsError, ClientError
import os

app = Flask(__name__)
app.secret_key = "your_secret_key_here"

# Load environment variables
LIARA_ENDPOINT_URL = os.getenv("LIARA_ENDPOINT_URL")
LIARA_ACCESS_KEY   = os.getenv("LIARA_ACCESS_KEY")
LIARA_SECRET_KEY   = os.getenv("LIARA_SECRET_KEY")
BUCKET_NAME        = os.getenv("BUCKET_NAME")

def upload_file(file, bucket_name):
  try:
      s3_client.upload_fileobj(file, bucket_name, file.filename)
      flash(f"File '{file.filename}' uploaded successfully.", "success")
  except ClientError as e:
      flash(f"Error uploading file: {e}", "error")

def generate_permanent_url(bucket_name, file_name):
  return f"{LIARA_ENDPOINT_URL}/{bucket_name}/{file_name}"

def list_files(bucket_name):
  try:
      response = s3_client.list_objects_v2(Bucket=bucket_name)
      return [obj['Key'] for obj in response.get('Contents', [])]
  except ClientError as e:
      flash(f"Error listing files: {e}", "error")
      return []

def generate_presigned_url(bucket_name, file_name, expiration=3600):
  try:
      url = s3_client.generate_presigned_url(
          'get_object',
          Params={'Bucket': bucket_name, 'Key': file_name},
          ExpiresIn=expiration
      )
      return url
  except ClientError as e:
      flash(f"Error generating pre-signed URL: {e}", "error")
      return None

import boto3
from botocore.config import Config
from urllib.parse import urlsplit, urlunsplit

CUSTOM_DOMAIN = "example.com"  # change this to your custom domain

presign_client = boto3.client(
  "s3",
  endpoint_url=LIARA_ENDPOINT_URL,
  aws_access_key_id=LIARA_ACCESS_KEY,
  aws_secret_access_key=LIARA_SECRET_KEY,
  config=Config(signature_version="s3v4", s3={"addressing_style": "virtual"}),
)

def generate_custom_domain_url(bucket_name, file_name, expiration=3600):
  signed = presign_client.generate_presigned_url(
      "get_object",
      Params={"Bucket": bucket_name, "Key": file_name},
      ExpiresIn=expiration,
  )
  parts = urlsplit(signed)
  return urlunsplit(parts._replace(netloc=CUSTOM_DOMAIN))

def delete_file(bucket_name, file_name):
  try:
      s3_client.delete_object(Bucket=bucket_name, Key=file_name)
      flash(f"File '{file_name}' deleted successfully.", "success")
  except ClientError as e:
      flash(f"Error deleting file: {e}", "error")

def list_buckets():
  try:
      response = s3_client.list_buckets()
      return [bucket['Name'] for bucket in response.get('Buckets', [])]
  except NoCredentialsError:
      flash("Credentials not available.", "error")
      return []
  except ClientError as e:
      flash(f"Error listing buckets: {e}", "error")
      return []

from flask import Flask, render_template, request, flash
import boto3
from botocore.exceptions import NoCredentialsError, ClientError
import os

app = Flask(__name__)
app.secret_key = "your_secret_key_here"

# Load environment variables
LIARA_ENDPOINT_URL = os.getenv("LIARA_ENDPOINT_URL")
LIARA_ACCESS_KEY   = os.getenv("LIARA_ACCESS_KEY")
LIARA_SECRET_KEY   = os.getenv("LIARA_SECRET_KEY")
BUCKET_NAME        = os.getenv("BUCKET_NAME")

# Initialize S3 client
s3_client = boto3.client(
  "s3",
  endpoint_url=LIARA_ENDPOINT_URL,
  aws_access_key_id=LIARA_ACCESS_KEY,
  aws_secret_access_key=LIARA_SECRET_KEY
)

# Controller: List all buckets
def list_buckets():
  try:
      response = s3_client.list_buckets()
      return [bucket['Name'] for bucket in response.get('Buckets', [])]
  except NoCredentialsError:
      flash("Credentials not available.", "error")
      return []
  except ClientError as e:
      flash(f"Error listing buckets: {e}", "error")
      return []

# Controller: List all files in a bucket
def list_files(bucket_name):
  try:
      response = s3_client.list_objects_v2(Bucket=bucket_name)
      return [obj['Key'] for obj in response.get('Contents', [])]
  except ClientError as e:
      flash(f"Error listing files: {e}", "error")
      return []

# Controller: Upload a file
def upload_file(file, bucket_name):
  try:
      s3_client.upload_fileobj(file, bucket_name, file.filename)
      flash(f"File '{file.filename}' uploaded successfully.", "success")
  except ClientError as e:
      flash(f"Error uploading file: {e}", "error")

# Controller: Delete a file
def delete_file(bucket_name, file_name):
  try:
      s3_client.delete_object(Bucket=bucket_name, Key=file_name)
      flash(f"File '{file_name}' deleted successfully.", "success")
  except ClientError as e:
      flash(f"Error deleting file: {e}", "error")

# Controller: Generate a pre-signed URL
def generate_presigned_url(bucket_name, file_name, expiration=3600):
  try:
      url = s3_client.generate_presigned_url(
          'get_object',
          Params={'Bucket': bucket_name, 'Key': file_name},
          ExpiresIn=expiration
      )
      return url
  except ClientError as e:
      flash(f"Error generating pre-signed URL: {e}", "error")
      return None

# Controller: Generate a permanent URL
def generate_permanent_url(bucket_name, file_name):
  return f"{LIARA_ENDPOINT_URL}/{bucket_name}/{file_name}"

# Main route
@app.route("/", methods=["GET", "POST"])
def index():
  if request.method == "POST":
      # Handle file upload
      if "file" in request.files:
          file = request.files["file"]
          if file.filename:
              upload_file(file, BUCKET_NAME)

      # Handle file deletion
      file_to_delete = request.form.get("delete_file")
      if file_to_delete:
          delete_file(BUCKET_NAME, file_to_delete)

  # Fetch bucket names and files
  buckets = list_buckets()
  files = list_files(BUCKET_NAME)

  # Generate pre-signed URLs and permanent URLs for each file
  presigned_urls = {}
  permanent_urls = {}
  for file in files:
      presigned_urls[file] = generate_presigned_url(BUCKET_NAME, file)
      permanent_urls[file] = generate_permanent_url(BUCKET_NAME, file)

  return render_template(
      "index.html",
      buckets=buckets,
      files=files,
      presigned_urls=presigned_urls,
      permanent_urls=permanent_urls
  )

# Run the app
if __name__ == "__main__":
  app.run(debug=True)

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>S3 File Manager</title>
  <style>
      body { font-family: Arial, sans-serif; margin: 20px; }
      .container { max-width: 800px; margin: auto; }
      .file-item { margin-bottom: 10px; display: flex; align-items: center; }
      .file-actions button { margin-left: 10px; }
      .message { margin-top: 20px; padding: 10px; border-radius: 5px; }
      .success { background-color: #d4edda; color: #155724; }
      .error { background-color: #f8d7da; color: #721c24; }
  </style>
</head>
<body>
  <div class="container">
      <h1>S3 File Manager</h1>

      <!-- Flash Messages -->
      {% with messages = get_flashed_messages(with_categories=true) %}
          {% if messages %}
              {% for category, message in messages %}
                  <div class="message {{ category }}">{{ message }}</div>
              {% endfor %}
          {% endif %}
      {% endwith %}

      <!-- Bucket Information -->
      <h2>Buckets</h2>
      <ul>
          {% for bucket in buckets %}
              <li>{{ bucket }}</li>
          {% endfor %}
      </ul>

      <!-- File Management -->
      <h2>Files in Bucket: {{ BUCKET_NAME }}</h2>
      <form method="POST" enctype="multipart/form-data">
          <input type="file" name="file" required>
          <button type="submit">Upload File</button>
      </form>

      <h3>File List</h3>
      {% if files %}
          <div>
              {% for file in files %}
                  <div class="file-item">
                      <span>{{ file }}</span>
                      <div class="file-actions">
                          <a href="{{ permanent_urls[file] }}" target="_blank">Permanent Link</a>
                          <a href="{{ presigned_urls[file] }}" target="_blank">Pre-signed Link</a>
                          <form method="POST" style="display: inline;">
                              <input type="hidden" name="delete_file" value="{{ file }}">
                              <button type="submit">Delete</button>
                          </form>
                      </div>
                  </div>
              {% endfor %}
          </div>
      {% else %}
          <p>No files found in the bucket.</p>
      {% endif %}
  </div>
</body>
</html>