QuotaGuard Heroku Django and Azure SQL DB

import pandas as pd
from PIL import Image
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
from dotenv import load_dotenv
import socks
import socket
import os
import io

load_dotenv()

CLIENT_ID = os.getenv("CLIENT_ID")
TENANT_ID = os.getenv("TENANT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")
DB_SERVER = os.getenv("DB_SERVER")
DB_NAME = os.getenv("DB_NAME")
DB_DRIVER = os.getenv("DB_DRIVER")
DB_TIMEOUT = os.getenv("DB_TIMEOUT")
QUOTAGUARD_URL = os.getenv("QUOTAGUARD_URL")

if QUOTAGUARD_URL:
    proxy_protocol, proxy_auth_host = QUOTAGUARD_URL.split("://")
    proxy_auth, proxy_host_port = proxy_auth_host.split("@")
    proxy_user, proxy_pass = proxy_auth.split(":")
    proxy_host, proxy_port = proxy_host_port.split(":")
    proxy_port = int(proxy_port)

socks.set_default_proxy(socks.SOCKS5, proxy_host, proxy_port, username=proxy_user, password=proxy_pass)
socket.socket = socks.socksocket

connection_string = (
    "mssql+pyodbc:///?odbc_connect=" +
    f"Driver={{{DB_DRIVER}}};" +
    f"Server={DB_SERVER};" +
    f"Database={DB_NAME};" +
    "Encrypt=yes;" +
    "TrustServerCertificate=yes;" +
    f"Connection Timeout={DB_TIMEOUT};" +
    "Authentication=ActiveDirectoryServicePrincipal;" +
    f"UID={CLIENT_ID};" +
    f"PWD={CLIENT_SECRET};" +
    f"Authority Id={TENANT_ID};"
)

engine = create_engine(connection_string, poolclass=NullPool)

try:
    query = "SELECT * FROM list_photos"
    with engine.connect() as conn:
        df = pd.read_sql(query, conn)
        print("DataFrame loaded successfully!")
        print(df.head())
except Exception as e:
    print(f"Error: {e}")

for index, row in df.iterrows():
    holdon = input("Press Enter to view the next image...")
    try:
        image = Image.open(io.BytesIO(row['photo']))
        image.show()
    except Exception as e:
        print(f"Error displaying image at index {index}: {e}")

I'm trying to pull the data from Azure SQL database via proxy but it is not using this proxy to connect to it therefore I'm getting an error from database where it says data source name not found etc etc etc which is typical when I'm trying to connect to DB via IP which is not in whitelist. The proxy's IP is in whitelist that's how I know it is not using proxy to make a request plust in QuotaGuard dashboard I don't see this request as well.

I've read all the documentation here: https://devcenter.heroku.com/articles/quotaguard#testing-in-the-python-interpreter

Also I've tried different methods of connecting, what am I doing wrong?

The below steps will help you implementation of QuotaGuard with Heroku, Django, and Azure SQL DB.

As a Prerequisites make sure you have the below tools:

  • pandas
  • PIL
  • sqlalchemy
  • create_engine for sqlalchemy.pool
  • NullPool for sqlalchemy.pool
  • dotenv
  • imports for socket and os

To create a new directory for your project and navigate into it:

mkdir myproject && cd myproject

To set up a virtual environment within the directory:

python3 -m venv venv

Activate the virtual environment:

source venv/bin/activate

You can install the required packages:

$ pip install django pandas Pillow sqlalchemy psycopg2-binary psycopg2-pool dotenv

Run the following command to start a new Django project:

django-admin startproject myproject

Create a new Azure sql database & Install the necessary packages to work with PostgreSQL

pip install psycopg2-binary psycopg2-pool

Create a .env file in the root directory of your project and include the below url:

DATABASE_URL=postgresql://username:password@your_server.database.windows.net:5432/mydatabase?sslmode=required

Next, add QuotaGuard to Your Django Project

In the myproject folder, create a new directory named quota_guard.

Add Middleware for Quota Handling

from django.http import HttpResponse
from django.utils.deprecation import MiddlewareMixin
from .quota import Quota
class QuotaMiddleware(MiddlewareMixin):
    Quota = Quota()
    def process_request(self, request):
        ip = request.META.get('REMOTE_ADDR')
        quota = self.Quota.get(ip)
        if not quota or quota.quota_exceeded():
            return HttpResponse(status=429)

Define the Quota Class:

import json
from collections import defaultdict
import functools
import ipaddress
from . import settings
class Quota:
    quota = defaultdict(int)
    @functools.lru_cache(maxsize=None)
    def get(ip):
        ip_address = ipaddress.ip_address(ip)
        return Quota.quota[ip_address]
    @functools.lru_cache(maxsize=None)
    def get_or_create(ip):
        quota = Quota.quota[ip]
        Quota.quota[ip] = Quota(ip, settings.RATE_LIMIT)
        return quota
class Quota:
    def __init__(self, ip, rate_limit):
        self.ip = ip
        self.rate_limit = rate_limit
        self.last_request = 0
        self.requests = 0
    def quota_exceeded(self):
        current_time = int(time.time())
        elapsed_time = current_time - self.last_request
        if elapsed_time < self.rate_limit:
            self.requests += 1
            self.last_request = current_time
            return self.requests > settings.MAX_REQUESTS

To test your QuotaGuard implementation, send requests to your Django application by running the following command:

gunicorn myproject.wsgi:application --bind 0.0.0.0:8000

Thank you @devcodef1 for the documentation on QuotaGuard: Implementing Heroku, Django, Azure SQL DB

Back to Top