Configuration and Engine Setup

Define connection parameters and construct the database URL string. Initialize the SQLAlchemy engine to handle connection pooling.

DB_HOST = '127.0.0.1'
DB_PORT = 3306
DB_USER = 'admin'
DB_PASS = 'secure_password'
DB_NAME = 'company_db'

CONNECTION_STRING = f'mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}'

from sqlalchemy import create_engine
db_engine = create_engine(CONNECTION_STRING, echo=True)

To execute raw SQL commands, establish a connection directly from the engine.

connection = db_engine.connect()
query_result = connection.execute('SELECT * FROM employees')
print(query_result.fetchall())
connection.close()

ORM Model Definition and Session Management

Set up the declarative base and configure the session factory for ORM operations.

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker

ORMBase = declarative_base()
SessionFactory = sessionmaker(bind=db_engine)
db_session = SessionFactory()

Define the table structure using a class inheriting from the base.

class Staff(ORMBase):
    __tablename__ = 'staff_records'
    
    emp_id = Column(Integer, primary_key=True, autoincrement=True)
    full_name = Column(String(100), nullable=False)
    work_years = Column(Integer)
    department = Column(String(50))

ORMBase.metadata.create_all(db_engine)

Data Insertion Operations

Instantiate model objects and persist them using the session.

new_employee = Staff(full_name='Alice Smith', work_years=5, department='Engineering')
db_session.add(new_employee)
db_session.commit()

For multiple records, utilize the bulk add method.

bulk_employees = [
    Staff(full_name='Bob Jones', work_years=2, department='Sales'),
    Staff(full_name='Carol White', work_years=7, department='HR')
]
db_session.add_all(bulk_employees)
db_session.commit()

Querying Records

Retrieve all entries from the table.

all_staff = db_session.query(Staff).all()
for person in all_staff:
    print(f"{person.full_name} - {person.department}")

Select specific columns instead of entire objects.

names_only = db_session.query(Staff.full_name).all()
first_name = db_session.query(Staff.full_name).first()

Apply filtering logic to narrow down results.

# Filter by condition
experienced_staff = db_session.query(Staff).filter(Staff.work_years >= 3).all()

# Sorting results
sorted_staff = db_session.query(Staff.full_name, Staff.work_years).order_by(Staff.work_years.desc()).all()

# Multiple AND conditions
specific_role = db_session.query(Staff).filter(Staff.work_years >= 3, Staff.department == 'Engineering').all()

Complex logical operations require explicit operators.

from sqlalchemy import or_

# OR condition
mixed_criteria = db_session.query(Staff).filter(
    or_(Staff.work_years >= 10, Staff.department == 'HR')
).all()

# Equality and Inequality
exact_match = db_session.query(Staff).filter(Staff.work_years == 5).all()
not_match = db_session.query(Staff).filter(Staff.work_years != 5).all()

# Pattern matching (LIKE)
like_query = db_session.query(Staff).filter(Staff.full_name.like('%Smith%')).all()

# IN clause
in_query = db_session.query(Staff).filter(Staff.work_years.in_([2, 7])).all()

# Count total records
total_count = db_session.query(Staff).count()

Updatnig Existing Entries

Modify record attributes using the update method followed by a commit.

rows_updated = db_session.query(Staff).filter(Staff.full_name == 'Bob Jones').update({'work_years': 3})
db_session.commit()

Removing Records

Delete entries matching specific criteria.

db_session.query(Staff).filter(Staff.full_name == 'Bob Jones').delete()
db_session.commit()

remaining_staff = db_session.query(Staff.full_name, Staff.work_years).all()