docs: Add example

Author: QuisEgoSum

example/__init__.py

@@ -0,0 +1,26 @@
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.orm import DeclarativeBase
+
+from sqlalchemy_tx_context import SQLAlchemyTransactionContext
+
+connect_args = dict(
+    max_cached_statement_lifetime=0,
+    statement_cache_size=5000,
+    server_settings={
+        'application_name': 'test'
+    }
+)
+
+
+engine = create_async_engine(
+    'postgresql+asyncpg://test:[email protected]:5432/test',
+    connect_args=connect_args,
+    pool_size=1,
+    max_overflow=0
+)
+
+db = SQLAlchemyTransactionContext(engine)
+
+
+class Base(DeclarativeBase):
+    __abstract__ = True

example/connection.py

@@ -0,0 +1,14 @@
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from example.connection import Base
+
+
+class Test(Base):
+    __tablename__ = 'test'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(36), unique=True)
+
+    def get_id(self):
+        return self.id

example/models.py

@@ -0,0 +1,34 @@
+import asyncio
+
+from example.connection import db
+from example.models import Test
+
+
+async def test():
+    delete_result = await db.delete(Test).rowcount()
+
+    print('Delete result', delete_result)
+
+    insert_result = await db.insert(Test).values(id=1, name='test 1').execute()
+    print(
+        'Inserted count',
+        insert_result.rowcount,
+        type(insert_result)
+    )
+
+    select_result = await db.select(Test.__table__).where(Test.id == 1).mapped_all()
+    print('Select all', select_result)
+
+    update_result = await db.update(Test)\
+        .where(Test.id == 1)\
+        .values(name='test 2')\
+        .returning(Test.__table__)\
+        .mapped_one()
+    print('Updated dict', dict(**update_result))
+
+
+if __name__ == '__main__':
+    asyncio.run(test())
+
+
+

example/queries.py

@@ -140,6 +140,5 @@ async def get_current_transaction(self) -> typing.ContextManager[AsyncSession]:
 
     def _proxy_sqlalchemy_query_factory(self, method: typing.Any) -> typing.Any:
         def wrapper(*args, **kwargs):
-            query = method(*args, **kwargs)
-            return ProxyQuery(query, self)
+            return ProxyQuery(method(*args, **kwargs), self)
         return wrapper