You're Invited:Meet the Socket Team at BlackHat and DEF CON in Las Vegas, Aug 4-6.RSVP โ†’
Socket
Book a DemoInstallSign in
Socket

dkbotzdb

Package Overview
Dependencies
Maintainers
1
Alerts
File Explorer

Advanced tools

Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

dkbotzdb

DkBotzDB Is The Most Powerful And Developer-Friendly Database Solution Powered By DKBotz And DkBotzPro, Offering Fast, Reliable, Scalable, And Secure Data Management. Designed For Effortless Integration And Built With Modern API Standards, DkBotzDB Ensures Smooth Performance Across Applications, Backed By The Trust And Excellence of dkbotzpro.in.

1.9.0
pipPyPI
Maintainers
1

DKBOTZDB

Welcome to DKBOTZDB! ๐Ÿš€

DKBOTZDB Is The Most Powerful And Developer-Friendly Database Solution Powered By DKBOTZ And DKBOTZPro, Offering Fast, Reliable, Scalable, And Secure Data Management. Designed For Effortless Integration And Built With Modern API Standards, dkbotzdb Ensures Smooth Performance Across Applications, Backed By The Trust And Excellence of dkbotzpro.in.

Table of Contents

Features

  • Quick and Robust Database Interaction: Insert, find, update and delete records easily.
  • Smart Find & Advanced Queries: Powerful query feature with regex support, range filters, sorting, limiting, and skipping results.
  • Case Insensitive: Recognizes multiple variations of class names ( DKBOTZDB, DKBOTZDB, DKBOTZdb ).
  • Flexible Syntax: Multiple ways to set tokens and collections (bracket notation, attribute access).
  • API Driven: Driven by HTTP requests with automatic re-try logic and robust error handling.
  • Scalable: Supports large amounts of data with no reduction in performance.
  • Secure: Supports secure API calls with token-based authentication.
  • Colored Logging: Beautiful colored console output will enhance your debug experience.
  • Admin Features: Token management and plan administration tools.

๐ŸŒŸ Why DKBOTZDB?

After years of struggling with complex database setups, poor documentation, and unreliable services, we created dkbotzdb to solve real developer problems:

  • โšก Lightning Fast: Sub-millisecond response times with intelligent caching
  • ๐Ÿ”ง Zero Setup: Start coding immediately - no configuration files or complex setup
  • ๐Ÿ›ก๏ธ Rock Solid: Enterprise-grade reliability with automatic failover
  • ๐ŸŽฏ Developer First: Intuitive API design with beautiful error messages
  • ๐Ÿ“ˆ Infinitely Scalable: From prototype to production without breaking a sweat
  • ๐Ÿ’ฐ Fair Pricing: Pay for what you use, not what you might use

Why We Built This

We were fed up with databases that produce promises but deliver headaches. There were too many occasions where we were struggling against:

  • Complicated setup processes that took hours
  • Poor documentation that left us guessing
  • Unreliable connections that failed at the worst moments
  • Limited query capabilities that forced us to write workarounds
  • Expensive pricing models that didn't make sense for growing projects

So we decided to build something better. dkbotzdb is our answer to these problems.

What Makes It Special

Super Fast Performance

When we say fast, we mean fast. This is not seconds, we are taking about milliseconds. As such, we have optimized the query path, and built intelligent caching mechanisms to guarantee it will still be responsive under heavy load.

Simple Integration

Do you recall the last time you started using a new service and it just worked out-of-the-box? That's what we're aiming for with dkbotzdb; start coding with just a few lines of code. No configuration files or multi-step environment setup.

Bulletproof Reliability

We understand that your data matters. Therefore, we designed multiple layers of redundancy, automatic failover processes and backup processes to make sure that your data is safe and applications are online.

Flexible Query System

Sometimes you just need CRUD operations. Other times you want complex aggregations, regex matching and range filters. dkbotzdb makes everything easy to query. dkbotzdb has a powerful query system, but it is also easy enough for a beginner to figure out.

Developer Experience First

All new features written for dkbotzdb go through one simple filter: "Will this make a developer's life easier?" If no, the feature doesn't get shipped. That's why you will see colored console logging, intuitive method names, and detailed error messages throughout the entire library.

Installation

You can install dkbotzdb using pip:

pip install dkbotzdb

Manual Installation

You can also install DKBOTZDB by cloning this repository:

# Clone the repository
git clone https://github.com/DKBOTZPROJECT/DKBOTZDB.git
cd DKBOTZDB

# Install dependencies
pip install requests colorlog

# Or install from requirements.txt
pip install -r requirements.txt

Setup

  • Create an account on DKBOTZDB.
  • Obtain your API Token from the DKBOTZDB dashboard.
  • Choose or create a collection where you want to store data.

Quick Start

from dkbotzdb import DKBOTZDB

# Method 1: Bracket notation (recommended)
db = DKBOTZDB()['YOUR_TOKEN']['your_collection']

# Method 2: Attribute access
db = DKBOTZDB().YOUR_TOKEN.your_collection

# Method 3: Traditional initialization
db = DKBOTZDB(token='YOUR_TOKEN')
db.collection = 'your_collection'

# Insert your first document
result = db.insert_one({"name": "John", "age": 30, "city": "New York"})
print("Document inserted:", result)

# Find the document
user = db.find_one({"name": "John"})
print("Found user:", user)

Basic Usage

Different Class Name Variations

# All these work the same way
from dkbotzdb import DKBOTZDB, DkBotzDB, DKBOTZdb

db1 = DKBOTZDB()['token']['collection']
db2 = DkBotzDB()['token']['collection']
db3 = DKBOTZdb()['token']['collection']

Setting Token and Collection

from dkbotzdb import DKBOTZDB

# Method 1: Bracket notation
db = DKBOTZDB()['YOUR_TOKEN']['users']

# Method 2: Attribute access
db = DKBOTZDB().YOUR_TOKEN.users

# Method 3: Mixed approach
db = DKBOTZDB()['YOUR_TOKEN'].users
db = DKBOTZDB().YOUR_TOKEN['users']

# Method 4: Traditional way
db = DKBOTZDB(token='YOUR_TOKEN')
db.collection = 'users'

CRUD Operations

Insert Operations

insert_one() / insertOne()

Insert a single document into the collection.

# Basic insertion
user_data = {
    "name": "Alice Johnson",
    "email": "alice@example.com",
    "age": 28,
    "department": "Engineering",
    "skills": ["Python", "JavaScript", "React"],
    "salary": 75000,
    "joined_date": "2023-01-15"
}

result = db.insert_one(user_data)
print("Inserted:", result)

# Using alias
result = db.insertOne(user_data)

insert_many() / insertMany()

Insert multiple documents at once.

users_data = [
    {"name": "Bob Smith", "age": 32, "department": "Marketing", "salary": 65000},
    {"name": "Carol White", "age": 29, "department": "Sales", "salary": 60000},
    {"name": "David Brown", "age": 35, "department": "Engineering", "salary": 80000},
    {"name": "Eve Davis", "age": 26, "department": "HR", "salary": 55000}
]

result = db.insert_many(users_data)
print("Inserted multiple:", result)

# Using alias
result = db.insertMany(users_data)

Find Operations

find_one() / findOne()

Find a single document matching the query.

# Find by name
user = db.find_one({"name": "Alice Johnson"})
print("Found user:", user)

# Find by multiple criteria
engineer = db.find_one({"department": "Engineering", "age": {"$gte": 30}})
print("Found engineer:", engineer)

# Find without query (returns first document)
first_user = db.find_one()
print("First user:", first_user)

# Using alias
user = db.findOne({"name": "Bob Smith"})

find()

Find multiple documents matching the query.

# Find all engineers
engineers = db.find({"department": "Engineering"})
print("All engineers:", engineers)

# Find with age range
young_employees = db.find({"age": {"$lt": 30}})
print("Young employees:", young_employees)

# Find with multiple conditions
high_earners = db.find({
    "salary": {"$gte": 70000},
    "department": {"$in": ["Engineering", "Marketing"]}
})
print("High earners:", high_earners)

# Find with limit and skip
limited_results = db.find(
    query={"department": "Engineering"},
    limit=5,
    skip=2,
    sort={"salary": -1}  # Sort by salary descending
)
print("Limited results:", limited_results)

# Find all documents
all_users = db.find()
print("All users:", all_users)

Update Operations

update_one() / updateOne()

Update a single document.

# Update specific field
result = db.update_one(
    {"name": "Alice Johnson"},
    {"$set": {"age": 29, "salary": 78000}}
)
print("Update result:", result)

# Add new field
result = db.update_one(
    {"name": "Bob Smith"},
    {"$set": {"remote_work": True, "last_updated": "2024-01-15"}}
)

# Increment a value
result = db.update_one(
    {"name": "Carol White"},
    {"$inc": {"salary": 5000}}
)

# Using alias
result = db.updateOne({"name": "David Brown"}, {"$set": {"department": "DevOps"}})

update_many() / updateMany()

Update multiple documents.

# Update all engineers
result = db.update_many(
    {"department": "Engineering"},
    {"$set": {"remote_work": True, "updated": "2024-01-15"}}
)
print("Updated engineers:", result)

# Give raise to all employees with salary < 70000
result = db.update_many(
    {"salary": {"$lt": 70000}},
    {"$inc": {"salary": 5000}}
)
print("Salary updates:", result)

# Using alias
result = db.updateMany(
    {"age": {"$gte": 30}},
    {"$set": {"senior_employee": True}}
)

replace_one() / replaceOne()

Replace an entire document.

# Replace entire document
new_user_data = {
    "name": "Alice Johnson",
    "email": "alice.johnson@newcompany.com",
    "age": 29,
    "department": "Data Science",
    "skills": ["Python", "Machine Learning", "SQL"],
    "salary": 85000,
    "joined_date": "2024-01-01"
}

result = db.replace_one(
    {"name": "Alice Johnson"},
    new_user_data
)
print("Replace result:", result)

# Using alias
result = db.replaceOne({"name": "Bob Smith"}, new_user_data)

Delete Operations

delete_one() / deleteOne()

Delete a single document.

# Delete by name
result = db.delete_one({"name": "Eve Davis"})
print("Delete result:", result)

# Delete by multiple criteria
result = db.delete_one({
    "department": "HR",
    "age": {"$lt": 25}
})

# Using alias
result = db.deleteOne({"name": "Carol White"})

delete_many() / deleteMany()

Delete multiple documents.

# Delete all interns
result = db.delete_many({"department": "Intern"})
print("Deleted interns:", result)

# Delete employees with low salary
result = db.delete_many({"salary": {"$lt": 50000}})
print("Deleted low salary employees:", result)

# Delete by age range
result = db.delete_many({"age": {"$gte": 65}})

# Using alias
result = db.deleteMany({"active": False})

Advanced Operations

count_documents() / countDocuments() / count()

Count documents matching a query.

# Count all documents
total_count = db.count_documents()
print("Total employees:", total_count)

# Count engineers
engineer_count = db.count_documents({"department": "Engineering"})
print("Total engineers:", engineer_count)

# Count with complex query
high_earner_count = db.count_documents({
    "salary": {"$gte": 75000},
    "age": {"$lt": 40}
})
print("Young high earners:", high_earner_count)

# Using aliases
count1 = db.countDocuments({"department": "Sales"})
count2 = db.count({"remote_work": True})

aggregate() / smart_find()

Advanced querying with multiple filter options.

# Basic aggregate query
results = db.aggregate(
    query={"department": "Engineering"},
    sort={"salary": -1},
    limit=10
)
print("Top 10 engineers by salary:", results)

# Aggregate with regex search
results = db.aggregate(
    regex={"name": "^A.*"},  # Names starting with 'A'
    sort={"age": 1}
)
print("Names starting with A:", results)

# Aggregate with range filter
results = db.aggregate(
    range_filter={"salary": {"min": 60000, "max": 80000}},
    sort={"joined_date": -1}
)
print("Mid-range salaries:", results)

# Complex aggregate query
results = db.aggregate(
    query={"department": {"$in": ["Engineering", "Data Science"]}},
    regex={"skills": "Python"},
    range_filter={"age": {"min": 25, "max": 35}},
    sort={"salary": -1},
    limit=5,
    skip=2
)
print("Complex query results:", results)

# Count only mode
count = db.aggregate(
    query={"department": "Engineering"},
    count_only=True
)
print("Engineer count:", count)

# Using smart_find alias
results = db.smart_find(
    query={"remote_work": True},
    sort={"salary": -1},
    limit=20
)

distinct()

Get distinct values for a field.

# Get all unique departments
departments = db.distinct("department")
print("All departments:", departments)

# Get unique skills with filter
skills = db.distinct("skills", {"department": "Engineering"})
print("Engineering skills:", skills)

# Get unique ages
ages = db.distinct("age")
print("All ages:", ages)

# Get unique cities where employees live
cities = db.distinct("city", {"salary": {"$gte": 70000}})
print("Cities of high earners:", cities)

Collection Management

collections()

List all collections in your database.

# Get all collection names
all_collections = db.collections()
print("All collections:", all_collections)

# Example output: ['users', 'products', 'orders', 'logs']

drop()

Drop (delete) the current collection.

# Drop current collection
result = db.drop()
print("Collection dropped:", result)

# Note: This will permanently delete all data in the collection

drop_all() / dropall()

Drop all collections in the database.

# Drop all collections (use with extreme caution!)
result = db.drop_all()
print("All collections dropped:", result)

# Using alias
result = db.dropall()

# Warning: This will delete ALL your data permanently!

usage_info()

Get usage statistics for your database.

# Get usage information
usage = db.usage_info()
print("Usage stats:", usage)

# Example output might include:
# {
#   "total_collections": 5,
#   "total_documents": 1500,
#   "storage_used": "2.5MB",
#   "api_calls_today": 250,
#   "plan_limits": {...}
# }

Token & Plan Management

get_token_info() / token_info() / token_details()

Get information about a token.

# Get current token info
token_info = db.get_token_info()
print("Token info:", token_info)

# Check another token
other_token_info = db.get_token_info("other_token_here")
print("Other token info:", other_token_info)

# Using aliases
info1 = db.token_info()
info2 = db.token_details("some_token")

# Example output:
# {
#   "token_id": "abc123",
#   "name": "My App Token",
#   "plan": "Pro",
#   "expires": "2024-12-31",
#   "created": "2024-01-01",
#   "status": "active"
# }

plan_status() / check_plan() / user_plan()

Check plan status for a token.

# Check current token's plan
plan = db.plan_status()
print("Plan status:", plan)

# Check specific token's plan
plan = db.plan_status("user_token_here")
print("User plan:", plan)

# Using aliases
plan1 = db.check_plan()
plan2 = db.user_plan("another_token")

# Example output:
# {
#   "plan_name": "Pro",
#   "expires": "2024-12-31",
#   "api_calls_remaining": 9750,
#   "storage_limit": "100GB",
#   "collections_limit": 50
# }

list_plans() / plans() / get_plans()

List all available plans.

# Get all available plans
available_plans = db.list_plans()
print("Available plans:", available_plans)

# Using aliases
plans1 = db.plans()
plans2 = db.get_plans()

# Example output:
# [
#   {
#     "plan_id": "free",
#     "name": "Free",
#     "price": 0,
#     "api_calls": 1000,
#     "storage": "10MB"
#   },
#   {
#     "plan_id": "pro",
#     "name": "Pro",
#     "price": 29,
#     "api_calls": 100000,
#     "storage": "100GB"
#   }
# ]

Admin Operations

Note: These operations require admin privileges

generate_token() / create_token() / admin_token()

Generate a new token (admin only).

# Generate new token
new_token = db.generate_token(
    name="New User Token",
    plan_id="pro",
    duration_days=365
)
print("New token created:", new_token)

# Using aliases
token1 = db.create_token("API Token", "free", 30)
token2 = db.admin_token("Admin Token", "enterprise", 730)

# Example output:
# {
#   "token": "new_generated_token_here",
#   "name": "New User Token",
#   "plan": "pro",
#   "expires": "2025-01-15"
# }

activate_plan()

Activate a plan for a user (admin only).

# Activate plan for user
result = db.activate_plan(
    user_token="user_token_here",
    plan_id="pro",
    duration_days=365
)
print("Plan activated:", result)

# Activate different plan
result = db.activate_plan(
    user_token="another_user_token",
    plan_id="enterprise",
    duration_days=730
)

Method Aliases

dkbotzdb provides multiple aliases for the same operations to match different coding styles:

Primary MethodAliases
insert_one()insertOne()
insert_many()insertMany()
find_one()findOne()
update_one()updateOne()
update_many()updateMany()
delete_one()deleteOne()
delete_many()deleteMany()
replace_one()replaceOne()
count_documents()countDocuments(), count()
smart_find()aggregate()
drop_all()dropall()
get_token_info()token_info(), token_details()
plan_status()check_plan(), user_plan()
list_plans()plans(), get_plans()
generate_token()create_token(), admin_token()

Logging

dkbotzdb uses beautiful colored logging for better debugging experience:

  • ๐Ÿ”ต DEBUG (Cyan): Detailed debugging information
  • ๐ŸŸข INFO (Green): General information about operations
  • ๐ŸŸก WARNING (Yellow): Warning messages
  • ๐Ÿ”ด ERROR (Red): Error messages
  • ๐Ÿ”ด CRITICAL (Bold Red): Critical errors

Customizing Log Level

import logging
from dkbotzdb import logger

# Set different log levels
logger.setLevel(logging.DEBUG)    # Show all messages
logger.setLevel(logging.WARNING)  # Show only warnings and errors
logger.setLevel(logging.ERROR)    # Show only errors

Error Handling

dkbotzdb includes robust error handling with automatic retry logic:

from dkbotzdb import DKBOTZDB

db = DKBOTZDB()['YOUR_TOKEN']['test_collection']

# The library automatically handles:
# - Network timeouts (retries up to 3 times)
# - Connection errors (with exponential backoff)
# - Invalid JSON responses
# - API rate limits
# - Server errors

# Always check results
result = db.insert_one({"test": "data"})
if result:
    print("Success:", result)
else:
    print("Operation failed - check logs for details")

# For critical operations, you might want additional checks
user = db.find_one({"id": "important_user"})
if user is None:
    print("User not found or error occurred")
    # Handle the error case
else:
    print("User found:", user)

Complete Example

Here's a comprehensive example showing various operations:

from dkbotzdb import DKBOTZDB
import json

# Initialize database
db = DKBOTZDB()['YOUR_TOKEN']['employees']

# Sample data
employees = [
    {
        "id": 1,
        "name": "John Doe",
        "email": "john@company.com",
        "department": "Engineering",
        "position": "Senior Developer",
        "salary": 85000,
        "skills": ["Python", "JavaScript", "Docker"],
        "remote": True,
        "joined": "2022-03-15"
    },
    {
        "id": 2,
        "name": "Jane Smith",
        "email": "jane@company.com",
        "department": "Marketing",
        "position": "Marketing Manager",
        "salary": 72000,
        "skills": ["SEO", "Content Marketing", "Analytics"],
        "remote": False,
        "joined": "2021-08-22"
    },
    {
        "id": 3,
        "name": "Mike Johnson",
        "email": "mike@company.com",
        "department": "Engineering",
        "position": "DevOps Engineer",
        "salary": 90000,
        "skills": ["AWS", "Kubernetes", "Python"],
        "remote": True,
        "joined": "2023-01-10"
    }
]

def main():
    # Insert employees
    print("=== Inserting Employees ===")
    result = db.insert_many(employees)
    print(f"Inserted: {result}")

    # Count total employees
    print("\n=== Counting Employees ===")
    total = db.count_documents()
    print(f"Total employees: {total}")

    # Find all engineers
    print("\n=== Finding Engineers ===")
    engineers = db.find({"department": "Engineering"})
    print(f"Engineers: {json.dumps(engineers, indent=2)}")

    # Find remote workers
    print("\n=== Finding Remote Workers ===")
    remote_workers = db.smart_find(
        query={"remote": True},
        sort={"salary": -1}
    )
    print(f"Remote workers: {json.dumps(remote_workers, indent=2)}")

    # Update salary
    print("\n=== Updating Salary ===")
    update_result = db.update_one(
        {"name": "John Doe"},
        {"$inc": {"salary": 5000}}
    )
    print(f"Salary update: {update_result}")

    # Get distinct departments
    print("\n=== Distinct Departments ===")
    departments = db.distinct("department")
    print(f"Departments: {departments}")

    # Advanced search
    print("\n=== Advanced Search ===")
    high_earners = db.aggregate(
        query={"salary": {"$gte": 80000}},
        regex={"skills": "Python"},
        sort={"salary": -1}
    )
    print(f"High-earning Python developers: {json.dumps(high_earners, indent=2)}")

    # Get usage info
    print("\n=== Usage Info ===")
    usage = db.usage_info()
    print(f"Usage: {json.dumps(usage, indent=2)}")

if __name__ == "__main__":
    main()

Real-World Example

Let me show you how this all comes together in a real application:

from dkbotzdb import DKBOTZDB
import json
from datetime import datetime

# Initialize our database
db = DKBOTZDB()['YOUR_TOKEN']['employee_management']

class EmployeeManager:
    def __init__(self):
        self.db = db

    def add_employee(self, employee_data):
        """Add a new employee to the system"""
        # Add timestamp
        employee_data['created_at'] = datetime.now().isoformat()
        employee_data['status'] = 'active'

        result = self.db.insert_one(employee_data)
        if result:
            print(f"โœ… Successfully added {employee_data['name']}")
            return result
        else:
            print("โŒ Failed to add employee")
            return None

    def find_employees_by_department(self, department):
        """Get all employees in a specific department"""
        employees = self.db.find({"department": department, "status": "active"})
        return employees or []

    def promote_employee(self, employee_id, new_position, new_salary):
        """Promote an employee"""
        update_data = {
            "$set": {
                "position": new_position,
                "salary": new_salary,
                "last_promotion": datetime.now().isoformat()
            }
        }

        result = self.db.update_one({"employee_id": employee_id}, update_data)
        if result:
            print(f"๐ŸŽ‰ Employee {employee_id} promoted to {new_position}")
        return result

    def get_department_stats(self):
        """Get statistics about each department"""
        departments = self.db.distinct("department")
        stats = {}

        for dept in departments:
            count = self.db.count_documents({"department": dept, "status": "active"})
            avg_salary = self.db.aggregate(
                query={"department": dept, "status": "active"},
                count_only=False
            )

            stats[dept] = {
                "employee_count": count,
                "employees": avg_salary
            }

        return stats

    def search_employees(self, search_term):
        """Smart search across multiple fields"""
        return self.db.smart_find(
            regex={
                "$or": [
                    {"name": f".*{search_term}.*"},
                    {"email": f".*{search_term}.*"},
                    {"skills": f".*{search_term}.*"}
                ]
            },
            sort={"name": 1}
        )

# Example usage
if __name__ == "__main__":
    manager = EmployeeManager()

    # Add a new employee
    new_employee = {
        "employee_id": "EMP001",
        "name": "Jennifer Walsh",
        "email": "jennifer@company.com",
        "department": "Engineering",
        "position": "Software Developer",
        "salary": 75000,
        "skills": ["Python", "React", "PostgreSQL"],
        "remote_work": True
    }

    manager.add_employee(new_employee)

    # Find all engineers
    engineers = manager.find_employees_by_department("Engineering")
    print(f"Engineers on the team: {len(engineers)}")

    # Promote someone
    manager.promote_employee("EMP001", "Senior Software Developer", 85000)

    # Get department statistics
    stats = manager.get_department_stats()
    print("Department statistics:", json.dumps(stats, indent=2))

    # Search for someone
    results = manager.search_employees("Jennifer")
    print("Search results:", results)

๐Ÿ“Š Feature Comparison Matrix

FeaturedkbotzdbMongoDBFirebaseSupabase
Setup Time30 seconds30 minutes5 minutes10 minutes
Learning CurveMinimalSteepModerateModerate
PricingPay-as-you-goComplex tiersLimited freeLimited free
Real-timeโœ… (Coming Q2)โœ…โœ…โœ…
ACID Transactionsโœ… (Coming Q3)โœ…Limitedโœ…
GraphQLโœ… (Coming Q2)Manual setupโŒโœ…
Multi-languageโœ… (Coming Q1 '25)โœ…โœ…โœ…
Serverlessโœ… (Coming Q1 '25)โŒโœ…โœ…
Developer Experienceโญโญโญโญโญโญโญโญโญโญโญโญโญโญโญโญ

๐ŸŽฏ Use Cases & Industries

๐Ÿข Enterprise Applications

  • Employee management systems
  • Customer relationship management
  • Inventory tracking
  • Financial record keeping
  • Compliance reporting

๐Ÿš€ Startups & SMBs

  • MVP development
  • User authentication
  • Content management
  • Analytics tracking
  • A/B testing data

๐ŸŽฎ Gaming

  • Player profiles
  • Leaderboards
  • Game state storage
  • Analytics tracking
  • Real-time chat

๐Ÿ›’ E-commerce

  • Product catalogs
  • Order management
  • Customer profiles
  • Inventory tracking
  • Recommendation engines

๐Ÿ“ฑ Mobile Apps

  • User preferences
  • Offline sync
  • Push notification data
  • Analytics events
  • User-generated content

๐ŸŽ“ Education

  • Student records
  • Course management
  • Assignment tracking
  • Grade books
  • Learning analytics

๐Ÿ”— Integration Examples

Flask Web Application

from flask import Flask, request, jsonify
from dkbotzdb import DKBOTZDB

app = Flask(__name__)
db = DKBOTZDB()['YOUR_TOKEN']['users']

@app.route('/users', methods=['GET'])
def get_users():
    users = db.find()
    return jsonify(users or [])

@app.route('/users', methods=['POST'])
def create_user():
    user_data = request.json
    result = db.insert_one(user_data)
    if result:
        return jsonify({"success": True, "data": result}), 201
    return jsonify({"error": "Failed to create user"}), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

FastAPI Application

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dkbotzdb import DKBOTZDB

app = FastAPI()
db = DKBOTZDB()['YOUR_TOKEN']['products']

class Product(BaseModel):
    name: str
    price: float
    category: str

@app.get("/products")
async def get_products():
    products = db.find()
    return products or []

@app.post("/products")
async def create_product(product: Product):
    result = db.insert_one(product.dict())
    if result:
        return {"success": True, "data": result}
    raise HTTPException(status_code=400, detail="Failed to create product")

Discord Bot Integration

import discord
from discord.ext import commands
from dkbotzdb import DKBOTZDB

bot = commands.Bot(command_prefix='!')
db = DKBOTZDB()['YOUR_TOKEN']['discord_users']

@bot.event
async def on_ready():
    print(f'{bot.user} has connected to Discord!')

@bot.command(name='register')
async def register_user(ctx, *, name):
    user_data = {
        "discord_id": str(ctx.author.id),
        "username": str(ctx.author),
        "display_name": name,
        "server_id": str(ctx.guild.id)
    }

    result = db.insert_one(user_data)
    if result:
        await ctx.send(f"โœ… {name} registered successfully!")
    else:
        await ctx.send("โŒ Registration failed!")

bot.run('YOUR_BOT_TOKEN')

๐Ÿ›ก๏ธ Security Best Practices

Token Management

import os
from dotenv import load_dotenv

# Use environment variables
load_dotenv()
DB_TOKEN = os.getenv('DKBOTZDB_TOKEN')

# Never hardcode tokens
# โŒ Bad
db = DKBOTZDB()['hardcoded_token_123']['collection']

# โœ… Good
db = DKBOTZDB()[DB_TOKEN]['collection']

Data Sanitization

import re
import html

def sanitize_input(data):
    """Sanitize user input"""
    if isinstance(data, str):
        # Remove HTML tags
        data = html.escape(data)
        # Remove potential injection patterns
        data = re.sub(r'[<>"\';]', '', data)
    elif isinstance(data, dict):
        return {k: sanitize_input(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [sanitize_input(item) for item in data]

    return data

# Usage
safe_data = sanitize_input(user_input)
db.insert_one(safe_data)

Best Practices We Have Learned

After having used dkbotzdb in production on various projects, here's what we can share that will save you time:

  • Always Check Return Values: Network issues are commonplace. Always check that it worked before moving on.

  • Use Batch Operations: Rather than repeated calls to insert_one() fifty times, use insert_many() and pass it in fifty. It will be incredibly faster.

  • Consider Structure of Your Queries: Well structured queries will run faster and can often result in less api calls.

  • Monitor Your Usage: usage_info() is helpful to avoid surprises and monitor your api usage.

  • Build Error Handling & Retries: Your most critical operations should have retry logic.

  • Keep your tokens secure: Don't stick tokens in version control, use environment variables.

  • Use Descriptive Collection Names: user_profiles is better than users is better than data.

Best Practices

  • Always check return values - Operations may fail due to network issues or API limits
  • Use appropriate indexes - Structure your queries efficiently for better performance
  • Handle errors gracefully - Implement proper error handling in production code
  • Use batch operations - Prefer insert_many() over multiple insert_one() calls
  • Monitor usage - Keep track of your API usage with usage_info()
  • Secure your tokens - Never expose tokens in public repositories

Contributing

We welcome contributions to improve DKBOTZDB. To contribute:

  • Fork the repository on GitHub
  • Create a new feature branch (git checkout -b feature-name)
  • Make your changes with proper tests
  • Commit your changes (git commit -am 'Add new feature')
  • Push to the branch (git push origin feature-name)
  • Create a Pull Request

How to Suggest Features

  • Check existing GitHub Issues
  • Create a new issue with detailed description
  • Include use cases and examples
  • Be specific about expected behavior

๐ŸŽ“ Learning Resources

๐Ÿ“š Documentation

๐ŸŽฏ Tutorials

  • Beginner: Building Your First App with DKBOTZDB
  • Intermediate: Advanced Queries and Aggregations
  • Advanced: Scaling with DKBOTZDB
  • Expert: Custom Integrations and Extensions

๐Ÿš€ Ready to Get Started?

  • Sign up for free at db.dkbotzpro.in
  • Install the package: pip install DKBOTZdb
  • Build something amazing!
from dkbotzdb import DKBOTZDB

# Your journey starts here
db = DKBOTZDB()['YOUR_TOKEN']['my_first_collection']

# Welcome to the future of database management
welcome = {
    "message": "Welcome to DKBOTZDB!",
    "status": "Ready to build amazing things",
    "excited": True
}

result = db.insert_one(welcome)
print("๐ŸŽ‰ You're all set! Let's build something incredible together!")

Support

License

This project is licensed under the GNU Affero General Public License v3.0.

What this means:

  • โœ… Commercial use allowed
  • โœ… Modification allowed
  • โœ… Distribution allowed
  • โœ… Patent use allowed
  • โ— Disclose source required
  • โ— License and copyright notice required
  • โ— Same license required for derivatives

See the LICENSE file for complete details.

๐Ÿ—บ๏ธ Complete Roadmap 2025-2027

โœ… Phase 1: Q1 2025 (Completed)

  • โš™๏ธ Core CRUD operations
  • ๐Ÿง  Aggregation pipeline support
  • ๐Ÿ” Token-based authentication
  • ๐Ÿ”— Multiple connection protocols
  • ๐Ÿšจ Robust error handling
  • ๐ŸŽจ Beautiful colored logging
  • ๐Ÿงฉ Method aliases (flexible commands)
  • ๐Ÿ—‚๏ธ Collection & database management
  • ๐Ÿงพ Plan management system

๐Ÿ”„ Phase 2: Q2 2025 (In Progress)

  • ๐Ÿ”ด Real-time Subscriptions (Change streams)
  • ๐Ÿ” GraphQL API interface
  • ๐Ÿ“ค๐Ÿ“ฅ Data Export/Import in bulk
  • ๐Ÿ’พ Backup & Restore automation
  • ๐Ÿ›ก๏ธ Military-grade 3-layer encryption:
  • ๐Ÿ“Š Advanced analytics dashboard

โšก Phase 3: Q3โ€“Q4 2025 (Planned)

  • ๐Ÿ—ƒ๏ธ Multi-database support (PostgreSQL, MySQL adapters)
  • ๐Ÿ”„ ACID-compliant Transactions
  • ๐Ÿ”Ž Full-text search engine integration (Elastic)
  • ๐Ÿ—บ๏ธ Geospatial queries (location-based indexing)
  • โณ Time-series collection support
  • ๐Ÿ”— Relational references (joins, population)
  • โš™๏ธ Serverless function runner (event-based triggers)

๐ŸŒŸ Phase 4: 2026 (Scalability & Intelligence)

  • ๐Ÿค– Machine learning for query prediction
  • ๐Ÿ“ˆ Auto-scaling clusters
  • ๐Ÿ“ก Performance monitoring with alerts
  • ๐Ÿ‘ฅ Team & collaboration tools
  • ๐Ÿ“ฌ Webhooks & event management
  • ๐Ÿ› ๏ธ ETL/ELT pipelines for big data
  • ๐Ÿงฎ Advanced indexing (TTL, compound, partial)

๐Ÿš€ Phase 5: 2027 (Experience & Ecosystem)

  • ๐Ÿ’ฌ AI-assisted query builder (text-to-query)
  • ๐Ÿ–ผ๏ธ Visual query designer (drag-and-drop GUI)
  • ๐Ÿ“‰ Built-in data visualization & charting
  • ๐Ÿ“ฑ Mobile SDKs (React Native, Flutter)
  • ๐ŸŒ IoT data ingestion (real-time & batch)
  • ๐Ÿง™ Predictive query analytics

Made By DKBOTZ Team โค๏ธ

FAQs

Did you know?

Socket

Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.

Install

Related posts