Structuring python package

.gitignore

@@ -0,0 +1,16 @@
+# Environment
+.vscode/
+
+# Build
+__pycache__/
+venv/
+*.py[cod]
+
+# Distribution
+*.egg-info/
+
+# Testing
+.mypy_cache/
+.pytest_cache/
+.tox/
+.coverage

pyproject.toml

@@ -0,0 +1,42 @@
+[tool.poetry]
+name = "sqlalchemy-events"
+version = "0.1.0"
+description = "Helper for SQLAlchemy Events"
+readme = "README.md"
+homepage = "https://github.com/nsiregar/sqlalchemy-events"
+repository = "https://github.com/nsiregar/sqlalchemy-events"
+authors = ["Ngalim Siregar <[email protected]>"]
+license = "MIT"
+classifiers = [
+    'Development Status :: 3 - Alpha',
+    'Environment :: Web Environment',
+    'Intended Audience :: Developers',
+    'License :: OSI Approved :: MIT License',
+    'Operating System :: OS Independent',
+    'Programming Language :: Python :: 3.8',
+    'Topic :: Database',
+    'Topic :: Software Development :: Libraries :: Python Modules',
+    'Topic :: Utilities',
+]
+include = [
+    "LICENSE"
+]
+
+[tool.poetry.dependencies]
+python = "^3.8"
+SQLAlchemy = "^1.4.22"
+
+[tool.poetry.dev-dependencies]
+pytest = "^6.2.4"
+pytest-cov = "^2.12.1"
+
+[tool.pytest.ini_options]
+filterwarnings = [
+    "ignore:.*Binary.*:sqlalchemy.exc.SADeprecationWarning",
+    "ignore:.*Decimal.*:sqlalchemy.exc.SAWarning",
+    "ignore:.*Passing `info=*:DeprecationWarning",
+]
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"

src/sqlalchemy_events.py

@@ -0,0 +1,51 @@
+from functools import partial
+from sqlalchemy import event
+
+# EVENTS DOCS
+# http://docs.sqlalchemy.org/en/rel_1_1/core/event.html
+# ORM EVENTS DOCS
+# http://docs.sqlalchemy.org/en/rel_1_1/orm/events.html
+
+
+class _EventObject(object):
+    ATTR = "_sqlalchemy_event"
+
+    def __init__(self, field_name, event_name, listen_kwargs=None):
+        self.field_name = field_name
+        self.event_name = event_name
+        self.listen_kwargs = listen_kwargs or {}
+
+
+def listen_events(*args):
+    def wrapper(cls):
+        for name, fn in cls.__dict__.items():
+            if not name.startswith("__") and hasattr(fn, _EventObject.ATTR):
+                e = getattr(fn, _EventObject.ATTR)
+                if e.field_name:
+                    event.listen(
+                        getattr(cls, e.field_name), e.event_name, fn, **e.listen_kwargs
+                    )
+                else:
+                    event.listen(cls, e.event_name, fn, **e.listen_kwargs)
+        return cls
+
+    if args and callable(args[0]):
+        return wrapper(args[0])
+    return wrapper
+
+
+def on(*args, **listen_kwargs):
+    if len(args) == 1:
+        field_name, event_name = (None, args[0])
+    elif len(args) == 2:
+        field_name, event_name = args
+    else:
+        raise NotImplementedError("@on accepts only one or two positional arguments")
+
+    def wrapper(fn):
+        setattr(
+            fn, _EventObject.ATTR, _EventObject(field_name, event_name, listen_kwargs)
+        )
+        return fn
+
+    return wrapper

tests/attach_events_test.py

@@ -0,0 +1,28 @@
+from unittest import TestCase
+from sqlalchemy import event
+from src.sqlalchemy_events import listen_events, on
+
+
+class MockEvent(event.Events):
+    def mock_event_one(self):
+        pass
+
+    def mock_event_two(self):
+        pass
+
+
+class MockClass(object):
+    dispatch = event.dispatcher(MockEvent)
+
+
+class AttachEventsTest(TestCase):
+    def test_wrap_to_class(self):
+        @listen_events
+        class Target(MockClass):
+            @on("mock_event_one")
+            def sample_method(self):
+                pass
+
+        target = Target()
+        assert len(target.dispatch.mock_event_one) == 1
+        assert len(target.dispatch.mock_event_two) == 0