本文整理汇总了Python中sqlahelper.get_session函数的典型用法代码示例。如果您正苦于以下问题:Python get_session函数的具体用法?Python get_session怎么用?Python get_session使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_session函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _teardown_db
def _teardown_db():
import transaction
transaction.abort()
import sqlahelper
sqlahelper.get_session().remove()
from . import models as m
m.Base.metadata.drop_all()
开发者ID:rebeccaframework,项目名称:shirly,代码行数:7,代码来源:tests.py
示例2: test_add_engine_twice
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)
开发者ID:pyramid-collective,项目名称:SQLAHelper,代码行数:8,代码来源:tests.py
示例3: test_member
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)
开发者ID:rebeccaframework,项目名称:shirly,代码行数:13,代码来源:tests.py
示例4: setUp
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)
开发者ID:jbking,项目名称:pyramid_auth_demo,代码行数:15,代码来源:tests.py
示例5: test1
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)
开发者ID:pyramid-collective,项目名称:SQLAHelper,代码行数:32,代码来源:tests.py
示例6: stream_view
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}
)
开发者ID:yoshrote,项目名称:columns2,代码行数:29,代码来源:blog.py
示例7: find_object
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
开发者ID:jbking,项目名称:pyramid_auth_demo,代码行数:7,代码来源:resources.py
示例8: app_settings
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)
开发者ID:yoshrote,项目名称:columns2,代码行数:7,代码来源:view.py
示例9: __getitem__
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
开发者ID:yoshrote,项目名称:columns2,代码行数:28,代码来源:sqla.py
示例10: main
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)
开发者ID:geoportallux,项目名称:geoportailv3-gisgr,代码行数:59,代码来源:tooltips2pot.py
示例11: tearDown
def tearDown(self):
config.cleanup_system()
sm = self.p_config
sm.__init__('base')
testing.tearDown()
Session = sqlahelper.get_session()
Session.expunge_all()
开发者ID:blaflamme,项目名称:ptah,代码行数:8,代码来源:base.py
示例12: get_author_data_from_user
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 {}
开发者ID:yoshrote,项目名称:columns2,代码行数:8,代码来源:models.py
示例13: clear
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
开发者ID:yoshrote,项目名称:columns2,代码行数:8,代码来源:sqla.py
示例14: main_pages
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]
开发者ID:yoshrote,项目名称:columns2,代码行数:9,代码来源:helpers.py
示例15: db_session_request
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
开发者ID:yoshrote,项目名称:columns2,代码行数:9,代码来源:models.py
示例16: __init__
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()
开发者ID:geoportallux,项目名称:geoportailv3-gisgr,代码行数:9,代码来源:PF.py
示例17: add
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
开发者ID:yoshrote,项目名称:columns2,代码行数:11,代码来源:sqla.py
示例18: __setitem__
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
开发者ID:yoshrote,项目名称:columns2,代码行数:12,代码来源:sqla.py
示例19: test_multiple_engines
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)
开发者ID:pyramid-collective,项目名称:SQLAHelper,代码行数:12,代码来源:tests.py
示例20: test_multiple_engines_without_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())
开发者ID:pyramid-collective,项目名称:SQLAHelper,代码行数:12,代码来源:tests.py
注:本文中的sqlahelper.get_session函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论