Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
''' initialise temporary new database. '''
self.ctx = streetsign_server.app.test_request_context
# streetsign_server.app.config['MODE'] = 'testing'
models.bcrypt = MockBcrypt()
streetsign_server.app.config['DATABASE_FILE'] = ':memory:'
streetsign_server.app.config['TESTING'] = True
models.DB = SqliteDatabase(None)
models.DB.init(streetsign_server.app.config['DATABASE_FILE'])
model_list = []
for modelname in models.__all__:
model = getattr(models, modelname)
try:
model._meta.database = models.DB # pylint: disable=protected-access
model_list.append(model)
except AttributeError:
pass
models.DB.create_tables(model_list)
self.client = streetsign_server.app.test_client()
def test_database_creation(tmpdir):
db = peewee.SqliteDatabase(':memory:')
manager = DatabaseManager(db, directory=tmpdir)
assert isinstance(manager.database, peewee.SqliteDatabase)
db = {'engine': 'peewee.SqliteDatabase', 'name': ':memory:'}
manager = DatabaseManager(db, directory=tmpdir)
assert isinstance(manager.database, peewee.SqliteDatabase)
db = 'sqlite:///:memory:'
manager = DatabaseManager(db, directory=tmpdir)
assert isinstance(manager.database, peewee.SqliteDatabase)
def _db_cache() -> peewee.Database:
path = db_path()
crc = crc32(path.encode())
if __cache.get(crc, None) is None:
__cache[crc] = peewee.SqliteDatabase(path)
return __cache[crc]
def create_connection() -> SqliteDatabase:
"""
create a database connection
:rtype: SqliteDatabase
"""
global _db
if _db:
return _db
else:
logger.debug('create new db connection')
_db = SqliteDatabase(get_config('db_path', './scylla.db'))
return _db
import peewee
from bottle import static_file
from datetime import datetime
from ray.authentication import Authentication, register
from ray.hooks import DatabaseHook
from ray.wsgi.wsgi import application
from ray.endpoint import endpoint
from ray_peewee.all import PeeweeModel
from ray.actions import Action, action
from ray.shield import Shield
database = peewee.SqliteDatabase('example.db')
class DBModel(PeeweeModel):
class Meta:
database = database
class MailHelper(object):
@classmethod
def send_email(self, to, message):
# fake send email
print('sending email to: %s with message: %s' % (to, message))
class UserHook(DatabaseHook):
pass
except ImportError:
pass
class _PooledSqliteDatabase(PooledDatabase):
def _is_closed(self, key, conn):
closed = super(_PooledSqliteDatabase, self)._is_closed(key, conn)
if not closed:
try:
conn.total_changes
except:
return True
return closed
class PooledSqliteDatabase(_PooledSqliteDatabase, SqliteDatabase):
pass
try:
from playhouse.sqlite_ext import SqliteExtDatabase
class PooledSqliteExtDatabase(_PooledSqliteDatabase, SqliteExtDatabase):
pass
except ImportError:
pass
def test_get_database_with_defined_database_type(self):
self.assertIsInstance(get_database(env('DB_TYPE')), SqliteDatabase)
"""Defines how connections with databases via peewee will be handled"""
import hug
from peewee import MySQLDatabase, PostgresqlDatabase, SqliteDatabase, Model
from playhouse.berkeleydb import BerkeleyDatabase
from playhouse.shortcuts import model_to_dict
ENGINES = {'sqlite': SqliteDatabase, 'psql': PostgresqlDatabase, 'mysql': MySQLDatabase, 'berkeley': BerkeleyDatabase}
def manage(api, engine='sqlite', location=':memory:', **settings):
'''Creates a new database object for handling connections based on the specified parameters'''
if not engine in ENGINES:
raise ValueError(('{0} is not one of the registered database engines: {1}.'
"To add an engine: hug_peewee.ENGINES['{0}'] = Class").format(engine,
', '.join(ENGINES.keys())))
engine_instance = ENGINES[engine](location, **settings)
if location != ':memory:':
hug_api = hug.API(api)
@hug.request_middleware(api=hug_api)
def process_data(request, response):
engine_instance.connect()
@hug.response_middleware(api=hug_api)
"""database base"""
import os
import traceback
import os.path as path
from threading import RLock
import peewee as db
DB_PATH = "/tmp/data.db"
DATABASE = db.SqliteDatabase(DB_PATH)
class BaseModel(db.Model):
"""Base Model for all db"""
class Meta:
database = DATABASE
def create_tables(tables):
"""create a list of tables"""
L = RLock()
L.acquire()
try:
print("Trying to create a table")
DATABASE.create_tables(tables)
print("Table created successfully")
except:
print(traceback.format_exc())