init_appwrite.py

import os
import time
from dotenv import load_dotenv
from appwrite.client import Client
from appwrite.services.databases import Databases
from appwrite.id import ID
from appwrite.exception import AppwriteException

load_dotenv()

APPWRITE_ENDPOINT = os.getenv("APPWRITE_ENDPOINT")
APPWRITE_PROJECT_ID = os.getenv("APPWRITE_PROJECT_ID")
APPWRITE_API_KEY = os.getenv("APPWRITE_API_KEY")
APPWRITE_DATABASE_ID = os.getenv("APPWRITE_DATABASE_ID")

client = Client()
client.set_endpoint(APPWRITE_ENDPOINT)
client.set_project(APPWRITE_PROJECT_ID)
client.set_key(APPWRITE_API_KEY)

databases = Databases(client)

COLLECTIONS = {
    "users": [
        {"key": "email", "type": "string"},
        {"key": "hashed_password", "type": "string"},
        {"key": "full_name", "type": "string"}
    ],
    "students": [
        {"key": "name", "type": "string"},
        {"key": "email", "type": "string"},
        {"key": "hashed_password", "type": "string"},
        {"key": "mobile", "type": "string"},
        {"key": "dob", "type": "string"},
        {"key": "enrollment_date", "type": "string"},
        {"key": "course_id", "type": "integer"},
        {"key": "batch_id", "type": "integer"},
        {"key": "status", "type": "string"},
        {"key": "progress_percent", "type": "double"},
        {"key": "total_score", "type": "double"}
    ],
    "batches": [
        {"key": "name", "type": "string"},
        {"key": "course_id", "type": "integer"},
        {"key": "academic_year", "type": "string"},
        {"key": "start_date", "type": "string"},
        {"key": "end_date", "type": "string"},
        {"key": "max_students", "type": "integer"},
        {"key": "instructor", "type": "string"}
    ],
    "practice_cases": [
        {"key": "title", "type": "string"},
        {"key": "case_type", "type": "string"},
        {"key": "difficulty", "type": "string"},
        {"key": "student_id", "type": "string"},
        {"key": "batch_id", "type": "integer"},
        {"key": "status", "type": "string"},
        {"key": "score", "type": "double"}
    ],
    "gst_practice_cases": [
        {"key": "student_id", "type": "string"},
        {"key": "gstin", "type": "string"},
        {"key": "return_period", "type": "string"},
        {"key": "place_of_supply", "type": "string"},
        {"key": "transaction_type", "type": "string"},
        {"key": "total_taxable_value", "type": "double"},
        {"key": "total_cgst", "type": "double"},
        {"key": "total_sgst", "type": "double"},
        {"key": "total_igst", "type": "double"},
        {"key": "total_cess", "type": "double"},
        {"key": "validation_status", "type": "string"},
        {"key": "arn", "type": "string"}
    ],
    "gst_invoice_entries": [
        {"key": "gst_case_id", "type": "string"},
        {"key": "invoice_number", "type": "string"},
        {"key": "date", "type": "string"},
        {"key": "invoice_type", "type": "string"},
        {"key": "hsn", "type": "string"},
        {"key": "taxable_value", "type": "double"},
        {"key": "tax_rate", "type": "double"},
        {"key": "cgst_amount", "type": "double"},
        {"key": "sgst_amount", "type": "double"},
        {"key": "igst_amount", "type": "double"}
    ],
    "tds_practice_cases": [
        {"key": "student_id", "type": "string"},
        {"key": "deductor_tan", "type": "string"},
        {"key": "deductor_pan", "type": "string"},
        {"key": "financial_year", "type": "string"},
        {"key": "quarter", "type": "string"},
        {"key": "form_type", "type": "string"},
        {"key": "total_payment_amount", "type": "double"},
        {"key": "total_tds_deducted", "type": "double"},
        {"key": "validation_status", "type": "string"},
        {"key": "receipt_number", "type": "string"}
    ],
    "tds_deduction_entries": [
        {"key": "tds_case_id", "type": "string"},
        {"key": "deductee_pan", "type": "string"},
        {"key": "section_code", "type": "string"},
        {"key": "payment_amount", "type": "double"},
        {"key": "deduction_date", "type": "string"},
        {"key": "tax_rate", "type": "double"},
        {"key": "tds_amount", "type": "double"}
    ],
    "itr_practice_cases": [
        {"key": "student_id", "type": "string"},
        {"key": "pan", "type": "string"},
        {"key": "assessment_year", "type": "string"},
        {"key": "itr_type", "type": "string"},
        {"key": "gross_total_income", "type": "double"},
        {"key": "total_deductions", "type": "double"},
        {"key": "net_taxable_income", "type": "double"},
        {"key": "old_regime_tax", "type": "double"},
        {"key": "new_regime_tax", "type": "double"},
        {"key": "recommended_regime", "type": "string"},
        {"key": "validation_status", "type": "string"},
        {"key": "ack_number", "type": "string"}
    ],
    "itr_income_sources": [
        {"key": "itr_case_id", "type": "string"},
        {"key": "head_of_income", "type": "string"},
        {"key": "amount", "type": "double"}
    ],
    "itr_deductions": [
        {"key": "itr_case_id", "type": "string"},
        {"key": "section", "type": "string"},
        {"key": "amount_claimed", "type": "double"},
        {"key": "amount_eligible", "type": "double"}
    ]
}

def create_schema():
    print(f"Initializing Appwrite Schema in Database {APPWRITE_DATABASE_ID}...")
    for collection_name, attributes in COLLECTIONS.items():
        try:
            databases.create_collection(
                database_id=APPWRITE_DATABASE_ID,
                collection_id=collection_name,
                name=collection_name,
                permissions=[],
                document_security=True
            )
            print(f"Created collection: {collection_name}")
            time.sleep(1) # Sleep slightly so collection is registered before attributes
        except AppwriteException as e:
            if "already exists" in str(e).lower():
                print(f"Collection {collection_name} already exists.")
            else:
                print(f"Error creating collection {collection_name}: {e}")
                continue

        for attr in attributes:
            try:
                if attr["type"] == "string":
                    databases.create_string_attribute(
                        database_id=APPWRITE_DATABASE_ID,
                        collection_id=collection_name,
                        key=attr["key"],
                        size=255,
                        required=False
                    )
                elif attr["type"] == "integer":
                    databases.create_integer_attribute(
                        database_id=APPWRITE_DATABASE_ID,
                        collection_id=collection_name,
                        key=attr["key"],
                        required=False
                    )
                elif attr["type"] == "double":
                    databases.create_float_attribute(
                        database_id=APPWRITE_DATABASE_ID,
                        collection_id=collection_name,
                        key=attr["key"],
                        required=False
                    )
                print(f"  - Requested attribute: {attr['key']}")
                time.sleep(0.5) # Throttle to avoid rate limits
            except AppwriteException as e:
                if "already exists" in str(e).lower() or "limit" in str(e).lower():
                    pass
                else:
                    print(f"  - Error creating attribute {attr['key']}: {e}")

if __name__ == "__main__":
    create_schema()