Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_io_loop(self):
with self.assertRaises(TypeError):
motor.MotorClient(test.env.rs_uri, io_loop='foo')
def prepare_db_backend(settings):
"""
Create MongoDB connection, ensure indexes
"""
mongo = motor.MotorClient(
host=settings.get("host", "localhost"),
port=settings.get("port", 27017),
max_pool_size=settings.get("max_pool_size", 10)
).open_sync()[settings.get("name", "centrifuge")]
ensure_indexes(mongo)
return mongo
def __get_db(self, db_name=None):
"""
:param db_name:
:return:
"""
if db_name is None:
db_name = system.configuration.DB_NAME
self._client = motor.MotorClient(system.configuration.DB_HOST, system.configuration.DB_PORT)[db_name]
# TODO: RPC over HTTPS
# TODO: a static-url function to set long cache TTL on media URLs
# TODO: Nginx cache media
# TODO: sitemap.xml
if __name__ == "__main__":
define_options(opts)
opts.parse_command_line()
for handler in logging.getLogger().handlers:
if hasattr(handler, 'baseFilename'):
print 'Logging to', handler.baseFilename
break
db = motor.MotorClient(opts.mongo_uri).get_default_database()
loop = tornado.ioloop.IOLoop.current()
loop.run_sync(partial(cache.startup, db))
if opts.rebuild_indexes or opts.ensure_indexes:
ensure_indexes = partial(indexes.ensure_indexes,
db,
drop=opts.rebuild_indexes)
loop.run_sync(ensure_indexes)
this_dir = os.path.dirname(__file__)
application = application.get_application(this_dir, db, opts)
http_server = httpserver.HTTPServer(application, xheaders=True)
http_server.listen(opts.port)
msg = 'Listening on port %s' % opts.port
print msg
import __main__
import logging
import pymongo
from .errors import ExtraFieldError, ValidationError
ok_types = (pymongo.MongoClient, pymongo.MongoReplicaSetClient)
try:
import motor
ok_types += (motor.MotorClient,)
except ImportError as e:
pass
logging.basicConfig(format='[%(asctime)s] %(levelname)s [%(module)s.%(funcName)s():%(lineno)d] %(message)s')
logger = logging.getLogger(__file__)
def valid_client(client):
"""returns ``True`` if input is pymongo or motor client
or any client added with allow_client()"""
return isinstance(client, ok_types)
def allow_client(client_type):
"""Allows another type to act as client type.
Intended for using with mock clients."""
def upload_original(fileid, extension, static_path, bucket_id):
conn = S3Connection(settings.AWS_ACCESS_KEY, settings.AWS_SECRET_KEY)
bucket = conn.lookup(bucket_id) or conn.create_bucket(bucket_id, location=Location.EU)
db_connection = motor.MotorClient().open_sync()
db = db_connection[settings.DATABASE_NAME]
original = find_original(fileid, static_path, extension)
if original:
relative_path = original.replace(static_path, '')
k = Key(bucket)
k.key = relative_path
print "Uploading original", original
s = os.stat(original)[stat.ST_SIZE]
print "%.1fKb" % (s / 1024.)
# reduced because I'm a cheapskate
k.set_contents_from_filename(original, reduced_redundancy=True)
print "Original uploaded"
else:
print "Original can't be found", repr(original)
(r"/fortunes", FortuneHandler),
],
template_path="templates"
)
application.ui_modules = {}
if __name__ == "__main__":
uvloop.install()
tornado.options.parse_command_line()
server = tornado.httpserver.HTTPServer(application)
server.bind(options.port, backlog=options.backlog, reuse_port=True)
server.start(0)
ioloop = tornado.ioloop.IOLoop.instance()
db = motor.MotorClient(options.mongo, maxPoolSize=500).hello_world
ioloop.start()
#coding: utf-8
import os
import motor
settings = {}
settings['debug'] = False
settings['app_name'] = u'App website'
settings['template_path'] = os.path.join(os.path.dirname(__file__), "templates")
settings['static_path'] = os.path.join(os.path.dirname(__file__), "static")
settings['xsrf_cookies'] = True
settings['cookie_secret'] = "81o0T=="
settings['login_url'] = '/login'
settings['session_secret'] = '08091287&^(01'
settings['session_dir'] = '/home/ohmyrepo/tmp/session'
settings['db'] = motor.MotorClient('localhost', 27017).ohmyrepo
# github application settings
githubapi = {}
githubapi['CLIENT_ID'] =''
githubapi['CLIENT_SECRET'] = ''
githubapi['REDIRECT_URL'] = ''
githubapi['ACCESS_TOKEN_URL'] = ''
githubapi['STATE'] = ''
githubapi['WEBHOOK_URL'] = ''
githubapi['OAUTH_URL'] = ''
def __init__(self, mongo_uri, signals):
parsed = urlparse(mongo_uri)
mongo = motor.MotorClient(parsed.netloc, parsed.port)
self.db = mongo[parsed.path.lstrip('/')]
self.col = self.db.sites
self.signals = signals