dm-dbcore

from dm_dbcore import DatabaseConnection, session_scope
from sqlalchemy import text

# Create connection (first time only, required on first call)
db = DatabaseConnection(
    database_connection_string='postgresql+psycopg://user:pass@localhost/mydb',
    cache_name='myapp_cache.pkl'  # Optional: enables metadata caching
)

# Subsequent calls return the same instance (no parameters needed)
db = DatabaseConnection()

# Use the connection with a transactional scope
with session_scope(db) as session:
    result = session.execute(text("SELECT * FROM users"))
    for row in result:
        print(row)

from dm_dbcore import DatabaseConnection, DBTYPE_POSTGRESQL, DBTYPE_MYSQL, DBTYPE_SQLITE

# PostgreSQL
db = DatabaseConnection('postgresql+psycopg://user:pass@localhost/mydb')

# MySQL
db = DatabaseConnection('mysql://user:pass@localhost/mydb')

# SQLite
db = DatabaseConnection('sqlite:///path/to/database.db')

# Check database type
print(db.database_type)  # 'postgresql', 'mysql', or 'sqlite'

from dm_dbcore import DatabaseConnection, session_scope
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# Create connection with metadata
db = DatabaseConnection('postgresql+psycopg://user:pass@localhost/mydb')

# Bind models to the connection's metadata
Base.metadata.bind = db.engine

# Query using ORM
with session_scope(db) as session:
    users = session.query(User).filter(User.name.like('John%')).all()
    for user in users:
        print(f"{user.name}: {user.email}")

# Enable caching by providing a cache filename
db = DatabaseConnection(
    database_connection_string='postgresql+psycopg://localhost/mydb',
    cache_name='myapp_metadata.pkl'
)

# Cache is automatically stored in ~/.sqlalchemy_cache/
# Cache is invalidated automatically when schema changes are detected

from dm_dbcore import DatabaseConnection
from contextlib import contextmanager

db = DatabaseConnection('postgresql+psycopg://localhost/mydb')

@contextmanager
def my_session():
    session = db.Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Use your custom context manager
with my_session() as session:
    # Your database operations
    pass

from dm_dbcore import DatabaseConnection
import numpy as np

db = DatabaseConnection('postgresql+psycopg://localhost/mydb')
# NumPy adapters are automatically loaded for PostgreSQL

# Arrays are automatically converted to/from database format

from dm_dbcore.mysql import read_password_from_my_cnf, read_connection_options_from_my_cnf

# Read password for specific host/user
password = read_password_from_my_cnf(host='localhost', user='myuser')

# Read all connection options from .my.cnf
options = read_connection_options_from_my_cnf(section='client')
# Returns: {'host': '...', 'user': '...', 'password': '...', 'database': '...', 'port': ...}

from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import base as pg
from dm_dbcore.adapters import PGPoint, PGPolygon

# Register types with SQLAlchemy
pg.ischema_names['point'] = PGPoint
pg.ischema_names['polygon'] = PGPolygon

# Use in your models
class Location(Base):
    __tablename__ = 'locations'
    id = Column(Integer, primary_key=True)
    coordinates = Column(PGPoint)  # Stores (x, y) tuples
    boundary = Column(PGPolygon)   # Stores list of (x, y) tuples

from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import base as pg
from dm_dbcore.adapters import PGASTCircle, PGASTPolygon

# Register astronomy types
pg.ischema_names['circle'] = PGASTCircle
pg.ischema_names['polygon'] = PGASTPolygon

# Use with astronomical coordinate systems
class AstronomicalObject(Base):
    __tablename__ = 'objects'
    id = Column(Integer, primary_key=True)
    search_region = Column(PGASTCircle)   # Uses cornish.ASTCircle
    footprint = Column(PGASTPolygon)      # Uses cornish.ASTPolygon

# Core functionality
from dm_dbcore import DatabaseConnection, session_scope
from dm_dbcore import DBTYPE_POSTGRESQL, DBTYPE_MYSQL, DBTYPE_SQLITE

# PostgreSQL geometric types
from dm_dbcore.adapters import PGPoint, PGPolygon

# Astronomy types (requires cornish)
from dm_dbcore.adapters import PGASTCircle, PGASTPolygon

# MySQL utilities
from dm_dbcore.mysql import read_password_from_my_cnf
from dm_dbcore.mysql import read_connection_options_from_my_cnf

with session_scope(db) as session:
    # Your database operations
    pass  # Automatic commit on success, rollback on exception