← Back to branch summary

~ubuntu-branches/ubuntu/vivid/ceilometer/vivid

« back to all changes in this revision

Viewing changes to ceilometer/openstack/common/db/sqlalchemy/utils.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2014-03-06 14:44:28 UTC
  • mto: (28.1.1 utopic-proposed) (1.2.1)
  • mto: This revision was merged to the branch mainline in revision 19.
  • Revision ID: package-import@ubuntu.com-20140306144428-rvphsh4igwyulzf0
Tags: upstream-2014.1~b3
Import upstream version 2014.1~b3

Show diffs side-by-side

added added

removed removed

Lines of Context:
ceilometer/openstack/common/db/sqlalchemy/utils.py
16
16
#    License for the specific language governing permissions and limitations
17
17
#    under the License.
18
18
 
 
19
import logging
19
20
import re
20
21
 
21
22
from migrate.changeset import UniqueConstraint
37
38
from sqlalchemy.types import NullType
38
39
 
39
40
from ceilometer.openstack.common.gettextutils import _
40
 
 
41
 
from ceilometer.openstack.common import log as logging
42
41
from ceilometer.openstack.common import timeutils
43
42
 
44
43
 
497
496
        where(new_table.c.deleted == deleted).\
498
497
        values(deleted=default_deleted_value).\
499
498
        execute()
 
499
 
 
500
 
 
501
def get_connect_string(backend, database, user=None, passwd=None):
 
502
    """Get database connection
 
503
 
 
504
    Try to get a connection with a very specific set of values, if we get
 
505
    these then we'll run the tests, otherwise they are skipped
 
506
    """
 
507
    args = {'backend': backend,
 
508
            'user': user,
 
509
            'passwd': passwd,
 
510
            'database': database}
 
511
    if backend == 'sqlite':
 
512
        template = '%(backend)s:///%(database)s'
 
513
    else:
 
514
        template = "%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s"
 
515
    return template % args
 
516
 
 
517
 
 
518
def is_backend_avail(backend, database, user=None, passwd=None):
 
519
    try:
 
520
        connect_uri = get_connect_string(backend=backend,
 
521
                                         database=database,
 
522
                                         user=user,
 
523
                                         passwd=passwd)
 
524
        engine = sqlalchemy.create_engine(connect_uri)
 
525
        connection = engine.connect()
 
526
    except Exception:
 
527
        # intentionally catch all to handle exceptions even if we don't
 
528
        # have any backend code loaded.
 
529
        return False
 
530
    else:
 
531
        connection.close()
 
532
        engine.dispose()
 
533
        return True
 
534
 
 
535
 
 
536
def get_db_connection_info(conn_pieces):
 
537
    database = conn_pieces.path.strip('/')
 
538
    loc_pieces = conn_pieces.netloc.split('@')
 
539
    host = loc_pieces[1]
 
540
 
 
541
    auth_pieces = loc_pieces[0].split(':')
 
542
    user = auth_pieces[0]
 
543
    password = ""
 
544
    if len(auth_pieces) > 1:
 
545
        password = auth_pieces[1].strip()
 
546
 
 
547
    return (user, password, database, host)