sqlalchemy

---

1. Create a new database and table:

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('postgresql://user:password@localhost:5432/my_database')
metadata = MetaData()
users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255), nullable=False),
    Column('email', String(255), nullable=False))
metadata.create_all(engine)

2. Insert data into a table:

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
new_user = User(name='John Doe', email='[e-mail]')
session.add(new_user)
session.commit()

3. Query data from a table:

users = session.query(User).all()
for user in users:
    print(user.name, user.email)

4. Update data in a table:

user = session.query(User).filter_by(id=1).first()
user.name = 'Jane Doe'
session.commit()

5. Delete data from a table:

session.delete(user)
session.commit()

6. Join tables:

from sqlalchemy.orm import aliased
user_alias = aliased(User)
orders = session.query(User, user_alias.name).join(user_alias, User.id == user_alias.id).all()

7. Group by:

orders = session.query(User.name, func.count(Order.id)).group_by(User.name).all()

8. Order by:

orders = session.query(User.name).order_by(User.name.desc()).all()

9. Limit results:

orders = session.query(User.name).limit(10).all()

10. Offset results:

orders = session.query(User.name).offset(10).all()

11. Use raw SQL:

from sqlalchemy import text
results = session.execute(text('SELECT * FROM users'))

12. Bulk operations:

from sqlalchemy.orm import load_only
results = session.query(User).options(load_only('name')).all()

13. Caching:

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine, expire_on_commit=False)
session = Session()

14. Relationships:

from sqlalchemy import ForeignKey
orders = Table('orders', metadata,
    Column('id', Integer, primary_key=True),
    Column('user_id', Integer, ForeignKey('users.id')))

15. Eager loading:

users = session.query(User).options(joinedload('orders')).all()

16. Lazy loading:

user = session.query(User).first()
user.orders  # Lazy load the orders

17. Pagination:

from sqlalchemy.orm import paginate
page = paginate(session.query(User), page=1, per_page=10)

18. Custom types:

from sqlalchemy.types import TypeDecorator
class MyType(TypeDecorator):
    impl = String

    def process_bind_param(self, value, dialect):
        return value.upper()

    def process_result_value(self, value, dialect):
        return value.lower()

19. Event listeners:

from sqlalchemy import event
@event.listens_for(User, 'after_insert')
def my_after_insert_listener(mapper, connection, target):
    print('New user inserted:', target)

20. Custom filters:

from sqlalchemy import func
@func.scalar
def my_custom_filter(user_id):
    return User.id == user_id

21. Custom validators:

from sqlalchemy.ext.declarative import declared_attr
from werkzeug.security import generate_password_hash, check_password_hash
class User:
    @declared_attr
    def password_hash(cls):
        return Column(String(128), nullable=False)
    def set_password(self, password):
        self.password_hash = generate_password_hash(password)
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

22. Custom query functions:

from sqlalchemy import func
@func.scalar
def my_custom_query_function(user_id):
    return session.query(User.name).filter_by(id=user_id).scalar()

23. Custom expression types:

from sqlalchemy import Column, Integer, expression
class MyExpressionType(expression.ColumnElement):
    def __init__(self, name):
        self.name = name

    def get_bind(self, compiler, **kw):
        return compiler.visit_column(self)

24. Custom connection processors:

from sqlalchemy.engine import Connection
def my_connection_processor(connection):
    connection.execute('SET TIME ZONE "Europe/London"')

25. Custom pool pre-pingers:

from sqlalchemy.engine import Engine
def my_pre_pinger(connection):
    connection.execute('SELECT 1')

26. Custom dialect:

from sqlalchemy.engine import default
class MyDialect(default.DefaultDialect):
    default_paramstyle = 'named'

27. Custom type coercion:

from sqlalchemy.types import TypeDecorator
class MyTypeCoercer(TypeDecorator):
    impl = String

    def coerce_result_value(self, value, dialect):
        return value.upper()

28. Custom dialect-specific compiler:

from sqlalchemy.engine.base import Compiled