SQLAlchemy Core — Type-Safe Queries Without the ORM

SQLAlchemy Core is a SQL expression language that compiles Python objects to dialect-correct SQL. Unlike the ORM, it maps directly to tables and columns — no models, no magic, just SQL built in Python with full type safety.

from sqlalchemy import (
    create_engine, MetaData, Table, Column, select, func,
    Integer, Text, Boolean, BigInteger, TIMESTAMP,
    and_, or_, case, cast, literal_column, text
)
from sqlalchemy.dialects.postgresql import aggregate_order_by
import pandas as pd

engine = create_engine(
    "postgresql+psycopg2://your_db_user:your_db_password"
    "@127.0.0.1/shared_db"
)

# ── Reflect existing tables ───────────────────────────────────────
meta = MetaData(schema="absolutelearning")
meta.reflect(bind=engine, only=["tw_users", "tw_tweets", "tw_hashtags",
                                 "tw_tweet_hashtags"])

users   = meta.tables["tw_users"]
tweets  = meta.tables["tw_tweets"]
tags    = meta.tables["tw_hashtags"]
tw_tags = meta.tables["tw_tweet_hashtags"]

# ── Simple SELECT ─────────────────────────────────────────────────
with engine.connect() as conn:
    stmt = (
        select(users.c.username, users.c.follower_count, users.c.verified)
        .where(users.c.verified == True)
        .order_by(users.c.follower_count.desc())
        .limit(10)
    )
    result = conn.execute(stmt)
    for row in result:
        print(row.username, row.follower_count)

# ── JOIN + aggregate ──────────────────────────────────────────────
with engine.connect() as conn:
    stmt = (
        select(
            users.c.username,
            users.c.verified,
            func.count(tweets.c.id).label("tweet_count"),
            func.sum(tweets.c.like_count).label("total_likes"),
            func.avg(tweets.c.like_count).label("avg_likes"),
        )
        .select_from(users.join(tweets, users.c.id == tweets.c.user_id))
        .where(
            and_(
                tweets.c.retweet_of_id == None,
                tweets.c.like_count > 10,
            )
        )
        .group_by(users.c.username, users.c.verified)
        .having(func.count(tweets.c.id) >= 5)
        .order_by(func.sum(tweets.c.like_count).desc())
        .limit(20)
    )
    df = pd.read_sql(stmt, conn)
    print(df)