# Django ORM Comparison - Know both frameworks
# Django model (contrast with SQLAlchemy)
from django.db import models

class Department(models.Model):
    name = models.CharField(max_length=50)

class Employee(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    department = models.ForeignKey(Department, on_delete=models.CASCADE)

# Django query (similar but different syntax)
Employee.objects.filter(department__name="HR").select_related('department')

# Async ORM - Modern Python requirement
# Requires SQLAlchemy 1.4+ and asyncpg
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=True,
)
async_session = AsyncSession(async_engine)

async with async_session.begin():
    result = await async_session.execute(
        select(Employee).where(Employee.name == "Alice")
    )
    employee = result.scalar_one()

# Testing Strategies - Interview differentiator
from unittest import mock

# Mock database for unit tests
with mock.patch('sqlalchemy.create_engine') as mock_engine:
    mock_conn = mock.MagicMock()
    mock_engine.return_value.connect.return_value = mock_conn
    
    # Test your ORM-dependent code
    create_employee("Test", "[email protected]")
    mock_conn.execute.assert_called()

# Production Monitoring - Track slow queries
from sqlalchemy import event

@event.listens_for(engine, "before_cursor_execute")
def before_cursor(conn, cursor, statement, params, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(engine, "after_cursor_execute")
def after_cursor(conn, cursor, statement, params, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop(-1)
    if total > 0.1:  # Log slow queries
        print(f"SLOW QUERY ({total:.2f}s): {statement}")

# Interview Power Move: Implement caching layer
from functools import lru_cache

class CachedEmployeeRepository(EmployeeRepository):
    @lru_cache(maxsize=100)
    def get_by_id(self, employee_id):
        return super().get_by_id(employee_id)
    
    def invalidate_cache(self, employee_id):
        self.get_by_id.cache_clear()

# Reduces database hits by 70% in read-heavy applications

# Pro Tip: Schema versioning in CI/CD pipelines
# Sample .gitlab-ci.yml snippet
deploy_db:
  stage: deploy
  script:
    - alembic upgrade head
    - pytest tests/db_tests.py  # Verify schema compatibility
  only:
    - main

# Real-World Case Study: E-commerce inventory system
class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    sku = Column(String(20), unique=True)
    stock = Column(Integer, default=0)
    
    # Atomic stock update (prevents race conditions)
    def decrement_stock(self, quantity, session):
        result = session.query(Product).filter(
            Product.id == self.id,
            Product.stock >= quantity
        ).update({"stock": Product.stock - quantity})
        if not result:
            raise ValueError("Insufficient stock")

# Usage during checkout
product.decrement_stock(2, session)

By: @DATASCIENCE4 🔒

#Python #ORM #SQLAlchemy #Django #Database #BackendDevelopment #CodingInterview #WebDevelopment #TechJobs #SystemDesign #SoftwareEngineering #DataEngineering #CareerGrowth #APIs #Microservices #DatabaseDesign #TechTips #DeveloperTools #Programming #CareerTips

#PyQt5 #SQLite #DesktopApp #WarehouseManagement #ERP #Python

Lesson: Advanced Warehouse ERP with PyQt5, SQLite, and Reporting

This tutorial covers building a complete desktop Enterprise Resource Planning (ERP) application for warehouse management. It features persistent storage using SQLite, inventory control, sales and purchase invoice management, a production module, and the ability to export reports to CSV.

---

#Step 1: Database Setup (database.py)

First, we create a dedicated file to handle all database interactions. This separation of concerns is crucial for a scalable application. Create a file named database.py.

import sqlite3
import csv

DB_NAME = 'warehouse.db'

def connect():
    return sqlite3.connect(DB_NAME)

def setup_database():
    conn = connect()
    cursor = conn.cursor()
    # Inventory Table: Stores raw materials and finished goods
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS inventory (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL UNIQUE,
            quantity INTEGER NOT NULL,
            price REAL NOT NULL
        )
    ''')
    # Invoices Table: Tracks both sales and purchases
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS invoices (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            type TEXT NOT NULL, -- 'SALE' or 'PURCHASE'
            party_name TEXT, -- Customer or Supplier Name
            date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    # Invoice Items Table: Links items from inventory to an invoice
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS invoice_items (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            invoice_id INTEGER,
            item_id INTEGER,
            quantity INTEGER NOT NULL,
            price_per_unit REAL NOT NULL,
            FOREIGN KEY (invoice_id) REFERENCES invoices (id),
            FOREIGN KEY (item_id) REFERENCES inventory (id)
        )
    ''')
    conn.commit()
    conn.close()

def get_inventory():
    conn = connect()
    cursor = conn.cursor()
    cursor.execute("SELECT id, name, quantity, price FROM inventory ORDER BY name")
    items = cursor.fetchall()
    conn.close()
    return items

def add_inventory_item(name, quantity, price):
    conn = connect()
    cursor = conn.cursor()
    try:
        cursor.execute("INSERT INTO inventory (name, quantity, price) VALUES (?, ?, ?)", (name, quantity, price))
        conn.commit()
    except sqlite3.IntegrityError:
        # Item with this name already exists
        return False
    finally:
        conn.close()
    return True

def update_item_quantity(item_id, change_in_quantity):
    conn = connect()
    cursor = conn.cursor()
    cursor.execute("UPDATE inventory SET quantity = quantity + ? WHERE id = ?", (change_in_quantity, item_id))
    conn.commit()
    conn.close()
    
def find_item_by_name(name):
    conn = connect()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM inventory WHERE name = ?", (name,))
    item = cursor.fetchone()
    conn.close()
    return item

# Add more functions here for invoices, etc. as we build the app.

# Hashtags: #SQLite #DatabaseDesign #DataPersistence #Python

---

#Step 2: Main Application Shell and Inventory Tab

Now, create the main application file, main.py. We'll build the window, tabs, and fully implement the Inventory tab, which will read from and write to our SQLite database.