oslo-db

import os from sqlalchemy import Column, Integer, String from oslo_db.sqlalchemy import enginefacade, models # Configure a simple SQLite database for demonstration os.environ['OSLO_DB_CONNECTION'] = 'sqlite:///./test.sqlite' # Initialize the enginefacade enginefacade.configure( sqlite_synchronous=False # For better performance in SQLite, though less safe ) # Define a base model for our application class MyModelBase(models.ModelBase, models.TimestampMixin): __abstract__ = True class User(MyModelBase): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(255), nullable=False) email = Column(String(255), unique=True, nullable=False) def __repr__(self): return f"<User(id='{self.id}', name='{self.name}', email='{self.email}')>" # Create tables (usually done via migration tools like Alembic in production) # For quickstart, we'll create directly if not exists with enginefacade.transaction() as session: MyModelBase.metadata.create_all(session.bind) # Example usage: adding and querying data class MyContext: pass @enginefacade.transaction_context_manager def add_user(context, name, email): new_user = User(name=name, email=email) context.session.add(new_user) print(f"Added user: {new_user}") return new_user @enginefacade.reader_context_manager def get_users(context): users = context.session.query(User).all() print("All users:") for user in users: print(f"- {user}") return users if __name__ == '__main__': # Use a dummy context object, as oslo_db often expects one ctx = MyContext() add_user(ctx, "Alice", "alice@example.com") add_user(ctx, "Bob", "bob@example.com") get_users(ctx) # Clean up test.sqlite if it exists if os.path.exists('./test.sqlite'): os.remove('./test.sqlite') print("Cleaned up test.sqlite")