Hello!
I try to use package and fail.
I have looked in #21 , #12
My orm classes are in one module orm.py.
import datetime
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy import (
create_engine,
text,
Column,
Integer,
Text,
ForeignKey,
LargeBinary,
Date,
DateTime,
Boolean
)
from sqlalchemy.orm import relationship, class_mapper, sessionmaker, exc
from sqlalchemy.dialects.postgresql import ENUM
from stubs import cyclop_pb2
from contextlib import contextmanager
import os
import importlib
from postgresql_audit import versioning_manager
import sqlalchemy as sa
# Enum
# A role of the user.
role = ('USER', 'ADMIN')
Role = ENUM(*role, name="role")
class Base(object):
@declared_attr
def __tablename__(cls):
"""Give every class a simple table name based on class name"""
return cls.__name__.lower()
id = Column(Integer, primary_key=True)
Base = declarative_base(cls = Base)
versioning_manager.init(Base)
class Users(Base):
"""Cyclop user.
Users:Sessions = 1:many.
A one to many bidirectional relationship of an Users to a Sessions.
A one to many relationship places a foreign key on the child
table referencing the parent.
Child will get a parent attribute with many-to-one semantics.
"""
__versioned__ = {}
login = Column(Text, nullable=False, unique=True)
password = Column(Text, nullable=False)
# A role of the user.
role = Column(Role, nullable=False)
alias = Column(Text)
archive = Column(Boolean, nullable=False)
sessions = relationship("Sessions", backref="users",
cascade="all, delete, delete-orphan", single_parent=True)
class Sessions(Base):
"""A session for client Users give the token.
Users:Sessions = 1:many.
"""
__versioned__ = {}
# A link to the one and only one Users instance.
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
token = Column(Text, nullable=False)
start_time = Column(DateTime, nullable=False)
end_time = Column(DateTime)
class ORM():
"""Provide a transactional scope context for session use.
To create db 'test' use as context:
with ORM('test') as orm:
orm.create_declarative_base(Base)
"""
def __init__(self, db='cyclop'):
"""Init don`t do much as __enter__() take role of context creation.
:param db: A database name.
"""
self.db = db
def create(self):
"""Create engine."""
self.engine = create_engine(
'postgresql://cyclop:cyclop@localhost/{0}'.format(self.db),
echo=True)
"""Create SQLAlchemy session."""
Session = sessionmaker(bind=self.engine)
self.session = Session()
sa.orm.configure_mappers() # <-- Important
# Removing these lines prevents audit trailing to work because the `audit_table` trigger is never attached.
tables = Base.metadata.tables
versioning_manager.audit_table(tables['users'])
Base.metadata.create_all(self.engine, checkfirst=True)
Unittest
from unittest import TestCase
import itertools
from datetime import datetime, timedelta, date
import orm
from stubs import cyclop_pb2
class TestCaseORM(TestCase):
"""DRY: To not write setUp 5 times."""
def setUp(self):
self.orm = orm.ORM('test')
self.orm.create()
self.session = self.orm.session
def tearDown(self):
self.orm.dispose()
class TestUsers(TestCaseORM):
def tearDown(self):
for somedata in self.session.query(orm.Users).filter(
orm.Users.login.ilike(u"%test%")).all():
self.session.delete(somedata)
self.session.commit()
super(self.__class__, self).tearDown()
def test_all_changes_by_id(self):
_user = orm.Users(login="test_user", password="12345", role="ADMIN",
archive = True, alias="alias")
try:
self.session.add(_user)
self.session.commit()
except:
self.session.rollback()
activities = self.session.query(versioning_manager.activity_cls).all()
counter = itertools.count()
for somedata in activities:
counter.next()
self.assertEqual(somedata.id, not None)
self.assertEqual(counter.next() > 1, True)
Output:
python -m unittest test_orm.TestUsers.test_all_changes_by_id
2017-09-14 10:43:34,489 INFO sqlalchemy.engine.base.Engine select version()
2017-09-14 10:43:34,489 INFO sqlalchemy.engine.base.Engine {}
2017-09-14 10:43:34,490 INFO sqlalchemy.engine.base.Engine select current_schema()
2017-09-14 10:43:34,490 INFO sqlalchemy.engine.base.Engine {}
2017-09-14 10:43:34,491 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2017-09-14 10:43:34,491 INFO sqlalchemy.engine.base.Engine {}
2017-09-14 10:43:34,491 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2017-09-14 10:43:34,491 INFO sqlalchemy.engine.base.Engine {}
2017-09-14 10:43:34,492 INFO sqlalchemy.engine.base.Engine show standard_conforming_strings
2017-09-14 10:43:34,492 INFO sqlalchemy.engine.base.Engine {}
2017-09-14 10:43:34,493 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-09-14 10:43:34,493 INFO sqlalchemy.engine.base.Engine {'name': u'transaction'}
2017-09-14 10:43:34,493 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-09-14 10:43:34,493 INFO sqlalchemy.engine.base.Engine {'name': u'users'}
2017-09-14 10:43:34,494 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-09-14 10:43:34,494 INFO sqlalchemy.engine.base.Engine {'name': u'sessions'}
2017-09-14 10:43:34,494 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-09-14 10:43:34,495 INFO sqlalchemy.engine.base.Engine {'name': u'persons'}
2017-09-14 10:43:34,495 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-09-14 10:43:34,495 INFO sqlalchemy.engine.base.Engine {'name': u'codes'}
2017-09-14 10:43:34,496 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-09-14 10:43:34,496 INFO sqlalchemy.engine.base.Engine {'name': u'activity'}
2017-09-14 10:43:34,496 INFO sqlalchemy.engine.base.Engine
SELECT EXISTS (
SELECT * FROM pg_catalog.pg_type t
WHERE t.typname = %(typname)s
AND pg_type_is_visible(t.oid)
)
2017-09-14 10:43:34,496 INFO sqlalchemy.engine.base.Engine {'typname': u'role'}
2017-09-14 10:43:34,497 INFO sqlalchemy.engine.base.Engine
SELECT EXISTS (
SELECT * FROM pg_catalog.pg_type t
WHERE t.typname = %(typname)s
AND pg_type_is_visible(t.oid)
)
2017-09-14 10:43:34,497 INFO sqlalchemy.engine.base.Engine {'typname': u'low_up'}
2017-09-14 10:43:34,497 INFO sqlalchemy.engine.base.Engine
SELECT EXISTS (
SELECT * FROM pg_catalog.pg_type t
WHERE t.typname = %(typname)s
AND pg_type_is_visible(t.oid)
)
2017-09-14 10:43:34,497 INFO sqlalchemy.engine.base.Engine {'typname': u'sex'}
2017-09-14 10:43:34,498 INFO sqlalchemy.engine.base.Engine
SELECT EXISTS (
SELECT * FROM pg_catalog.pg_type t
WHERE t.typname = %(typname)s
AND pg_type_is_visible(t.oid)
)
2017-09-14 10:43:34,498 INFO sqlalchemy.engine.base.Engine {'typname': u'person_status'}
2017-09-14 10:43:34,499 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2017-09-14 10:43:34,500 INFO sqlalchemy.engine.base.Engine INSERT INTO users (login, password, role, alias, archive) VALUES (%(login)s, %(password)s, %(role)s, %(alias)s, %(archive)s) RETURNING users.id
2017-09-14 10:43:34,500 INFO sqlalchemy.engine.base.Engine {'alias': 'alias', 'login': 'test_user', 'password': '12345', 'role': 'ADMIN', 'archive': True}
2017-09-14 10:43:34,501 INFO sqlalchemy.engine.base.Engine COMMIT
2017-09-14 10:43:34,503 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2017-09-14 10:43:34,504 INFO sqlalchemy.engine.base.Engine SELECT activity.id AS activity_id, activity.schema_name AS activity_schema_name, activity.table_name AS activity_table_name, activity.relid AS activity_relid, activity.issued_at AS activity_issued_at, activity.native_transaction_id AS activity_native_transaction_id, activity.verb AS activity_verb, activity.old_data AS activity_old_data, activity.changed_data AS activity_changed_data, activity.transaction_id AS activity_transaction_id
FROM activity
2017-09-14 10:43:34,504 INFO sqlalchemy.engine.base.Engine {}
F2017-09-14 10:43:34,505 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.login AS users_login, users.password AS users_password, users.role AS users_role, users.alias AS users_alias, users.archive AS users_archive
FROM users
WHERE users.login ILIKE %(login_1)s
2017-09-14 10:43:34,505 INFO sqlalchemy.engine.base.Engine {'login_1': u'%test%'}
2017-09-14 10:43:34,507 INFO sqlalchemy.engine.base.Engine SELECT sessions.id AS sessions_id, sessions.user_id AS sessions_user_id, sessions.token AS sessions_token, sessions.start_time AS sessions_start_time, sessions.end_time AS sessions_end_time
FROM sessions
WHERE %(param_1)s = sessions.user_id
2017-09-14 10:43:34,507 INFO sqlalchemy.engine.base.Engine {'param_1': 56}
2017-09-14 10:43:34,508 INFO sqlalchemy.engine.base.Engine DELETE FROM users WHERE users.id = %(id)s
2017-09-14 10:43:34,508 INFO sqlalchemy.engine.base.Engine {'id': 56}
2017-09-14 10:43:34,508 INFO sqlalchemy.engine.base.Engine COMMIT
======================================================================
FAIL: test_all_changes_by_id (test_orm.TestUsers)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test_orm.py", line 183, in test_all_changes_by_id
self.assertEqual(counter.next() > 1, True)
AssertionError: False != True
----------------------------------------------------------------------
Ran 1 test in 0.038s
FAILED (failures=1)
psql (9.6.4)
Type "help" for help.
test=> select * from activity;
id | schema_name | table_name | relid | issued_at | native_transaction_id | verb | old_data | changed_data | transaction_id
----+-------------+------------+-------+-----------+-----------------------+------+----------+--------------+----------------
(0 rows)
Hello! I try to use package and fail. I have looked in #21 , #12 My orm classes are in one module orm.py.
Unittest
Output: