def _teardown_db():
    import transaction
    transaction.abort()
    import sqlahelper
    sqlahelper.get_session().remove()
    from . import models as m
    m.Base.metadata.drop_all()

 def test_add_engine_twice(self):
     db1 = sa.create_engine(self.db1.url)
     db2 = sa.create_engine(self.db2.url)
     sqlahelper.add_engine(db1)
     self.assertIs(sqlahelper.get_session().bind, db1)
     sqlahelper.add_engine(db2)
     self.assertIs(sqlahelper.get_session().bind, db2)
     self.assertIs(sqlahelper.get_session().bind, sqlahelper.sessions.default.registry.registry.value.bind)

    def test_member(self):
        import sqlahelper
        from . import models as m
        p = m.Project(project_name=u"test-project")
        u = m.User(user_name=u'test-user')
        m = m.Member(project=p, user=u)
        sqlahelper.get_session().add(m)
        request = testing.DummyRequest(matchdict={'ticket_no': '10'},
            authenticated_user=u)
        target = self._makeOne(request)
        target.project = p

        self.assertEqual(target.member, m)

 def setUp(self):
     self.config = testing.setUp()
     from sqlalchemy import create_engine
     from sqlahelper import add_engine, get_session
     engine = create_engine('sqlite://')
     add_engine(engine)
     from .models import (
         Base,
         MyModel,
         )
     self.session = get_session()
     Base.metadata.create_all(engine)
     self.session = get_session()
     self.model = MyModel(name='one', value=55)
     self.session.add(self.model)

 def test1(self):
     import transaction
     Base = sqlahelper.get_base()
     class Person(Base):
         __tablename__ = "people"
         id = sa.Column(sa.Integer, primary_key=True)
         first_name = sa.Column(sa.Unicode(100), nullable=False)
         last_name = sa.Column(sa.Unicode(100), nullable=False)
     engine = sa.create_engine(self.db1.url)
     sqlahelper.add_engine(engine)
     Base.metadata.create_all()
     fred = Person(id=1, first_name=u"Fred", last_name=u"Flintstone")
     wilma = Person(id=2, first_name=u"Wilma", last_name=u"Flintstone")
     barney = Person(id=3, first_name=u"Barney", last_name=u"Rubble")
     betty = Person(id=4, first_name=u"Betty", last_name=u"Rubble")
     Session = sqlahelper.get_session()
     sess = Session()
     sess.add_all([fred, wilma, barney, betty])
     transaction.commit()
     sess.expunge_all()
     del fred, wilma, barney, betty
     # Can we get back a record?
     barney2 = sess.query(Person).get(3)
     self.assertEqual(barney2.id, 3)
     self.assertEqual(barney2.first_name, u"Barney")
     self.assertEqual(barney2.last_name, u"Rubble")
     sa.select([Person.first_name])
     # Can we iterate the first names in reverse alphabetical order?
     q = sess.query(Person.first_name).order_by(Person.first_name.desc())
     result = [x.first_name for x in q]
     control = [u"Wilma", u"Fred", u"Betty", u"Barney"]
     self.assertEqual(result, control)

def stream_view(request):
    tag = request.GET.get('tag')
    user = request.GET.get('user')
    page_number = request.GET.get('page')
    stories_per_page = 10
    try:
        page_number = max([int(page_number), 0])
    except (TypeError, ValueError):
        page_number = 0
    
    Session = sqlahelper.get_session()
    Session.rollback()
    query = Session.query(Article)
    if tag:
        query = query.filter(Article.tags.any(Tag.slug == tag))
    if user:
        query = query.filter(Article.author.has(User.name == user))
    
    stream = query.\
        filter(Article.published != None).\
        order_by(Article.published.desc()).\
        limit(stories_per_page).\
        offset(page_number * stories_per_page).\
        all()
    
    return render_to_response(
        'columns:templates/blog/stream.jinja', 
        {'stream': stream, 'request': request, 'page': page_number}
    )

def find_object(request):
    session = get_session()
    try:
        one = session.query(MyModel).filter(MyModel.name=='one').first()
    except DBAPIError:
        raise HTTPInternalServerError(explanation=conn_err_msg)
    return one

def app_settings(key, mod='core'):
    import sqlahelper
    from columns.models import Setting
    Session = sqlahelper.get_session()
    module = Session.query(Setting).get(mod)
    setting_dict = getattr(module, 'config', {})
    return setting_dict.get(key)

 def __getitem__(self, key):
     if key in self.collection_views:
         raise KeyError(key)
     if isinstance(key, slice):
         def is_negative(value):
             return value is not None and value < 0
         
         def offset_negative(length, value):
             try:
                 return length + value if int(value) <= 0 else value
             except TypeError:
                 return None
         
         if is_negative(key.start) or is_negative(key.stop):
             length = len(self)
             key = slice(offset_negative(length, key.start), offset_negative(length, key.stop))
         return self.index(offset=key.start, limit=key.stop)
     id_ = self._decode_key(key)
     db_session = sqlahelper.get_session()
     resource = db_session.query(
         self.__model__
     ).get(id_)
     if resource is None:
         raise KeyError(key)
         
     resource.__parent__ = self
     resource.__name__ = key
     return resource

def main():  # pragma: nocover
    env = bootstrap("development.ini")
    from geoportailv3.models import LuxGetfeatureDefinition

    package = env["registry"].settings["package"]
    directory = "%s/locale/" % package
    destination = path.join(directory, "%s-tooltips.pot" % package)

    w = codecs.open(destination, "wt", encoding="utf-8")
    w.write(
        u'''#, fuzzy
        msgid ""
        msgstr ""
        "MIME-Version: 1.0\\n"
        "Content-Type: text/plain; charset=utf-8\\n"
        "Content-Transfer-Encoding: 8bit\\n"
        '''
    )

    dbsession = sqlahelper.get_session()
    results = dbsession.query(LuxGetfeatureDefinition).\
        filter(LuxGetfeatureDefinition.remote_template == False).\
        filter(LuxGetfeatureDefinition.template == 'default_gisgr.html').all()  # noqa

    fields = []
    for result in results:
        engine = None
        first_row = None
        if result.query is not None and len(result.query) > 0:
            engine = sqlahelper.get_engine(result.engine)
            first_row = engine.execute("SELECT * FROM " + result.query).first()
        if result.rest_url is not None and len(result.rest_url) > 0:
            first_row = _get_external_data(
                result.rest_url,
                '96958.90059551848,61965.61097091329,' +
                '97454.77280739773,62463.21618929457', result.layer)

        if first_row is not None:
            attributes = dict(first_row)
            attributes = remove_attributes(
                attributes,
                result.attributes_to_remove,
                result.geometry_column)

            for attribute in attributes:
                if attribute not in fields:
                    fields.append(attribute)
                    w.write(
                        u'''#: engine:%(engine)s Layer:%(layer)s Role:%(role)s
        msgid "f_%(name)s"
        msgstr ""
        ''' % {
                            "engine": result.engine_gfi,
                            "layer": result.layer,
                            "role": result.role,
                            "name": attribute,
                        }
                    )
    print("tooltips Pot file updated: %s" % destination)

    def tearDown(self):
        config.cleanup_system()
        sm = self.p_config
        sm.__init__('base')
        testing.tearDown()

        Session = sqlahelper.get_session()
        Session.expunge_all()

def get_author_data_from_user():
    user_id = get_author_id()
    session = sqlahelper.get_session()
    try:
        user = session.query(User).get(user_id)
        return {'id':user_id, 'name':user.name, 'uri':user.profile}
    except: # pragma: no cover
        return {}

 def clear(self):
     db_session = sqlahelper.get_session()
     try:
         db_session.query(self.__model__).delete()
         db_session.commit()
     except: # pragma: no cover
         db_session.rollback()
         raise

def main_pages():
    import sqlahelper
    from columns.models import Page
    Session = sqlahelper.get_session()
    pages = Session.query(Page.slug, Page.title).\
        filter(Page.in_menu == True).\
        filter(Page.visible == True).\
        order_by(Page.in_main.desc(), Page.title.asc())
    return [(slug, title) for slug, title in pages]

def db_session_request(event):
    session = sqlahelper.get_session()
    def cleanup(_):
        try:
            session.rollback()
        except: # pragma: no cover
            pass
    event.request.add_finished_callback(cleanup)
    return session

 def __init__(self):
     self.client = \
         Client('https://titan.etat.lu/xxpfoWS/ParcelDetailVer1' +
                'Service/META-INF/wsdl/ParcelDetailVer1Service.wsdl')
     self.client_mesurage = \
         Client('https://titan.etat.lu/xxpfoWS/Measure' +
                'mentVer1Service/META-INF/wsdl/MeasurementVer1Service.wsdl')
     self.log = logging.getLogger(__name__)
     self.dbsession = sqlahelper.get_session()

 def add(self, resource):
     db_session = sqlahelper.get_session()
     try:
         saved_resource = db_session.merge(resource)
         db_session.commit()
     except: # pragma: no cover
         db_session.rollback()
         raise
     else:
         saved_resource.__name__ = saved_resource.get_key()
         return saved_resource

 def __setitem__(self, key, value):
     db_session = sqlahelper.get_session()
     value.set_key(key)
     try:
         saved_resource = db_session.merge(value)
         db_session.commit()
     except: # pragma: no cover
         db_session.rollback()
         raise
     else:
         saved_resource.__name__ = saved_resource.get_key()
         return saved_resource

 def test_multiple_engines(self):
     default = sa.create_engine(self.db1.url)
     stats = sa.create_engine(self.db2.url)
     sqlahelper.add_engine(default)
     sqlahelper.add_engine(stats, "stats")
     # Can we retrieve the engines?
     self.assertIs(sqlahelper.get_engine(), default)
     self.assertIs(sqlahelper.get_engine("default"), default)
     self.assertIs(sqlahelper.get_engine("stats"), stats)
     # Are the session binding and base binding set correctly?
     self.assertIs(sqlahelper.get_session().bind, default)
     self.assertIs(sqlahelper.get_base().metadata.bind, default)

 def test_multiple_engines_without_default(self):
     db1 = sa.create_engine(self.db1.url)
     db2 = sa.create_engine(self.db2.url)
     sqlahelper.add_engine(db1, "db1")
     sqlahelper.add_engine(db2, "db2")
     # Can we retrieve the engines?
     self.assertIs(sqlahelper.get_engine("db1"), db1)
     self.assertIs(sqlahelper.get_engine("db2"), db2)
     # There should be no default engine
     self.assertIsNone(sqlahelper.get_session().bind)
     self.assertIsNone(sqlahelper.get_base().metadata.bind)
     self.assertIsNone(sqlahelper.get_engine())