root/galaxy-central/eggs/SQLAlchemy-0.5.6_dev_r6498-py2.6.egg/sqlalchemy/engine/url.py

リビジョン 3, 6.2 KB (コミッタ: kohda, 14 年 前)

Install Unix tools  http://hannonlab.cshl.edu/galaxy_unix_tools/galaxy.html

行番号 
1"""Provides the :class:`~sqlalchemy.engine.url.URL` class which encapsulates
2information about a database connection specification.
3
4The URL object is created automatically when :func:`~sqlalchemy.engine.create_engine` is called
5with a string argument; alternatively, the URL is a public-facing construct which can
6be used directly and is also accepted directly by ``create_engine()``.
7"""
8
9import re, cgi, sys, urllib
10from sqlalchemy import exc
11
12
13class URL(object):
14    """
15    Represent the components of a URL used to connect to a database.
16
17    This object is suitable to be passed directly to a
18    ``create_engine()`` call.  The fields of the URL are parsed from a
19    string by the ``module-level make_url()`` function.  the string
20    format of the URL is an RFC-1738-style string.
21
22    All initialization parameters are available as public attributes.
23   
24    :param drivername: the name of the database backend. 
25      This name will correspond to a module in sqlalchemy/databases
26      or a third party plug-in.
27
28    :param username: The user name.
29
30    :param password: database password.
31
32    :param host: The name of the host.
33
34    :param port: The port number.
35
36    :param database: The database name.
37
38    :param query: A dictionary of options to be passed to the
39      dialect and/or the DBAPI upon connect.
40       
41    """
42
43    def __init__(self, drivername, username=None, password=None, host=None, port=None, database=None, query=None):
44        self.drivername = drivername
45        self.username = username
46        self.password = password
47        self.host = host
48        if port is not None:
49            self.port = int(port)
50        else:
51            self.port = None
52        self.database = database
53        self.query = query or {}
54
55    def __str__(self):
56        s = self.drivername + "://"
57        if self.username is not None:
58            s += self.username
59            if self.password is not None:
60                s += ':' + urllib.quote_plus(self.password)
61            s += "@"
62        if self.host is not None:
63            s += self.host
64        if self.port is not None:
65            s += ':' + str(self.port)
66        if self.database is not None:
67            s += '/' + self.database
68        if self.query:
69            keys = self.query.keys()
70            keys.sort()
71            s += '?' + "&".join("%s=%s" % (k, self.query[k]) for k in keys)
72        return s
73   
74    def __hash__(self):
75        return hash(str(self))
76   
77    def __eq__(self, other):
78        return \
79            isinstance(other, URL) and \
80            self.drivername == other.drivername and \
81            self.username == other.username and \
82            self.password == other.password and \
83            self.host == other.host and \
84            self.database == other.database and \
85            self.query == other.query
86           
87    def get_dialect(self):
88        """Return the SQLAlchemy database dialect class corresponding to this URL's driver name."""
89       
90        try:
91            module = getattr(__import__('sqlalchemy.databases.%s' % self.drivername).databases, self.drivername)
92            return module.dialect
93        except ImportError:
94            if sys.exc_info()[2].tb_next is None:
95                import pkg_resources
96                for res in pkg_resources.iter_entry_points('sqlalchemy.databases'):
97                    if res.name == self.drivername:
98                        return res.load()
99            raise
100 
101    def translate_connect_args(self, names=[], **kw):
102        """Translate url attributes into a dictionary of connection arguments.
103
104        Returns attributes of this url (`host`, `database`, `username`,
105        `password`, `port`) as a plain dictionary.  The attribute names are
106        used as the keys by default.  Unset or false attributes are omitted
107        from the final dictionary.
108
109        :param \**kw: Optional, alternate key names for url attributes.
110       
111        :param names: Deprecated.  Same purpose as the keyword-based alternate names,
112            but correlates the name to the original positionally.
113       
114        """
115
116        translated = {}
117        attribute_names = ['host', 'database', 'username', 'password', 'port']
118        for sname in attribute_names:
119            if names:
120                name = names.pop(0)
121            elif sname in kw:
122                name = kw[sname]
123            else:
124                name = sname
125            if name is not None and getattr(self, sname, False):
126                translated[name] = getattr(self, sname)
127        return translated
128
129def make_url(name_or_url):
130    """Given a string or unicode instance, produce a new URL instance.
131
132    The given string is parsed according to the RFC 1738 spec.  If an
133    existing URL object is passed, just returns the object.
134   
135    """
136    if isinstance(name_or_url, basestring):
137        return _parse_rfc1738_args(name_or_url)
138    else:
139        return name_or_url
140
141def _parse_rfc1738_args(name):
142    pattern = re.compile(r'''
143            (?P<name>\w+)://
144            (?:
145                (?P<username>[^:/]*)
146                (?::(?P<password>[^/]*))?
147            @)?
148            (?:
149                (?P<host>[^/:]*)
150                (?::(?P<port>[^/]*))?
151            )?
152            (?:/(?P<database>.*))?
153            '''
154            , re.X)
155
156    m = pattern.match(name)
157    if m is not None:
158        components = m.groupdict()
159        if components['database'] is not None:
160            tokens = components['database'].split('?', 2)
161            components['database'] = tokens[0]
162            query = (len(tokens) > 1 and dict(cgi.parse_qsl(tokens[1]))) or None
163            if query is not None:
164                query = dict((k.encode('ascii'), query[k]) for k in query)
165        else:
166            query = None
167        components['query'] = query
168
169        if components['password'] is not None:
170            components['password'] = urllib.unquote_plus(components['password'])
171
172        name = components.pop('name')
173        return URL(name, **components)
174    else:
175        raise exc.ArgumentError(
176            "Could not parse rfc1738 URL from string '%s'" % name)
177
178def _parse_keyvalue_args(name):
179    m = re.match( r'(\w+)://(.*)', name)
180    if m is not None:
181        (name, args) = m.group(1, 2)
182        opts = dict( cgi.parse_qsl( args ) )
183        return URL(name, *opts)
184    else:
185        return None
Note: リポジトリブラウザについてのヘルプは TracBrowser を参照してください。