# postgresql/pg8000.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors <see AUTHORS file>
+# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors <see AUTHORS
+# file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
.. dialect:: postgresql+pg8000
:name: pg8000
:dbapi: pg8000
- :connectstring: postgresql+pg8000://user:password@host:port/dbname[?key=value&key=value...]
+ :connectstring: \
+postgresql+pg8000://user:password@host:port/dbname[?key=value&key=value...]
:url: https://pythonhosted.org/pg8000/
Unicode
import decimal
from ... import processors
from ... import types as sqltypes
-from .base import PGDialect, \
- PGCompiler, PGIdentifierPreparer, PGExecutionContext,\
- _DECIMAL_TYPES, _FLOAT_TYPES, _INT_TYPES
+from .base import (
+ PGDialect, PGCompiler, PGIdentifierPreparer, PGExecutionContext,
+ _DECIMAL_TYPES, _FLOAT_TYPES, _INT_TYPES)
class _PGNumeric(sqltypes.Numeric):
if self.asdecimal:
if coltype in _FLOAT_TYPES:
return processors.to_decimal_processor_factory(
- decimal.Decimal,
- self._effective_decimal_return_scale)
+ decimal.Decimal, self._effective_decimal_return_scale)
elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
# pg8000 returns Decimal natively for 1700
return None
else:
raise exc.InvalidRequestError(
- "Unknown PG numeric type: %d" % coltype)
+ "Unknown PG numeric type: %d" % coltype)
else:
if coltype in _FLOAT_TYPES:
# pg8000 returns float natively for 701
return processors.to_float
else:
raise exc.InvalidRequestError(
- "Unknown PG numeric type: %d" % coltype)
+ "Unknown PG numeric type: %d" % coltype)
class _PGNumericNoBind(_PGNumeric):
class PGCompiler_pg8000(PGCompiler):
def visit_mod_binary(self, binary, operator, **kw):
return self.process(binary.left, **kw) + " %% " + \
- self.process(binary.right, **kw)
+ self.process(binary.right, **kw)
def post_process_text(self, text):
if '%%' in text:
def set_isolation_level(self, connection, level):
level = level.replace('_', ' ')
- print("level is", level)
- print("autocommit is", connection.autocommit)
- print("class is", connection)
if level == 'AUTOCOMMIT':
connection.connection.autocommit = True
"Valid isolation levels for %s are %s or AUTOCOMMIT" %
(level, self.name, ", ".join(self._isolation_lookup))
)
- print("autocommit is now", connection.autocommit)
dialect = PGDialect_pg8000
# coding: utf-8
-from sqlalchemy.testing.assertions import eq_, assert_raises, \
- assert_raises_message, AssertsExecutionResults, \
- AssertsCompiledSQL
+from sqlalchemy.testing.assertions import (
+ eq_, assert_raises, assert_raises_message, AssertsExecutionResults,
+ AssertsCompiledSQL)
from sqlalchemy.testing import engines, fixtures
from sqlalchemy import testing
import datetime
-from sqlalchemy import Table, Column, select, MetaData, text, Integer, \
- String, Sequence, ForeignKey, join, Numeric, \
- PrimaryKeyConstraint, DateTime, tuple_, Float, BigInteger, \
- func, literal_column, literal, bindparam, cast, extract, \
- SmallInteger, Enum, REAL, update, insert, Index, delete, \
- and_, Date, TypeDecorator, Time, Unicode, Interval, or_, Text
+from sqlalchemy import (
+ Table, Column, select, MetaData, text, Integer, String, Sequence, Numeric,
+ DateTime, BigInteger, func, extract, SmallInteger)
from sqlalchemy import exc, schema
from sqlalchemy.dialects.postgresql import base as postgresql
import logging
import logging.handlers
from sqlalchemy.testing.mock import Mock
-from sqlalchemy.engine.reflection import Inspector
+
class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
@testing.provide_metadata
def test_date_reflection(self):
metadata = self.metadata
- t1 = Table('pgdate', metadata, Column('date1',
- DateTime(timezone=True)), Column('date2',
- DateTime(timezone=False)))
+ Table(
+ 'pgdate', metadata, Column('date1', DateTime(timezone=True)),
+ Column('date2', DateTime(timezone=False)))
metadata.create_all()
m2 = MetaData(testing.db)
t2 = Table('pgdate', m2, autoload=True)
def mock_conn(res):
return Mock(
- execute=Mock(
- return_value=Mock(scalar=Mock(return_value=res))
- )
- )
-
- for string, version in \
- [('PostgreSQL 8.3.8 on i686-redhat-linux-gnu, compiled by '
- 'GCC gcc (GCC) 4.1.2 20070925 (Red Hat 4.1.2-33)', (8, 3,
- 8)),
- ('PostgreSQL 8.5devel on x86_64-unknown-linux-gnu, '
- 'compiled by GCC gcc (GCC) 4.4.2, 64-bit', (8, 5)),
- ('EnterpriseDB 9.1.2.2 on x86_64-unknown-linux-gnu, '
- 'compiled by gcc (GCC) 4.1.2 20080704 (Red Hat 4.1.2-50), '
- '64-bit', (9, 1, 2)),
- ('[PostgreSQL 9.2.4 ] VMware vFabric Postgres 9.2.4.0 '
- 'release build 1080137', (9, 2, 4))
-
- ]:
+ execute=Mock(return_value=Mock(scalar=Mock(return_value=res))))
+
+ for string, version in [
+ (
+ 'PostgreSQL 8.3.8 on i686-redhat-linux-gnu, compiled by '
+ 'GCC gcc (GCC) 4.1.2 20070925 (Red Hat 4.1.2-33)',
+ (8, 3, 8)),
+ (
+ 'PostgreSQL 8.5devel on x86_64-unknown-linux-gnu, '
+ 'compiled by GCC gcc (GCC) 4.4.2, 64-bit', (8, 5)),
+ (
+ 'EnterpriseDB 9.1.2.2 on x86_64-unknown-linux-gnu, '
+ 'compiled by gcc (GCC) 4.1.2 20080704 (Red Hat 4.1.2-50), '
+ '64-bit', (9, 1, 2)),
+ (
+ '[PostgreSQL 9.2.4 ] VMware vFabric Postgres 9.2.4.0 '
+ 'release build 1080137', (9, 2, 4))]:
eq_(testing.db.dialect._get_server_version_info(mock_conn(string)),
version)
def test_psycopg2_version(self):
v = testing.db.dialect.psycopg2_version
assert testing.db.dialect.dbapi.__version__.\
- startswith(".".join(str(x) for x in v))
+ startswith(".".join(str(x) for x in v))
# currently not passing with pg 9.3 that does not seem to generate
# any notices here, would rather find a way to mock this
else:
test_encoding = 'UTF8'
- e = engines.testing_engine(
- options={'client_encoding':test_encoding}
- )
+ e = engines.testing_engine(options={'client_encoding': test_encoding})
c = e.connect()
eq_(c.connection.connection.encoding, test_encoding)
def test_extract(self):
fivedaysago = datetime.datetime.now() \
- datetime.timedelta(days=5)
- for field, exp in ('year', fivedaysago.year), ('month',
- fivedaysago.month), ('day', fivedaysago.day):
+ for field, exp in ('year', fivedaysago.year), \
+ ('month', fivedaysago.month), ('day', fivedaysago.day):
r = testing.db.execute(select([extract(field, func.now()
+ datetime.timedelta(days=-5))])).scalar()
eq_(r, exp)
users.insert().execute(id=2, name='name2')
users.insert().execute(id=3, name='name3')
users.insert().execute(id=4, name='name4')
- eq_(users.select().where(users.c.name == 'name2'
- ).execute().fetchall(), [(2, 'name2')])
+ eq_(users.select().where(users.c.name == 'name2')
+ .execute().fetchall(), [(2, 'name2')])
eq_(users.select(use_labels=True).where(users.c.name
== 'name2').execute().fetchall(), [(2, 'name2')])
users.delete().where(users.c.id == 3).execute()
- eq_(users.select().where(users.c.name == 'name3'
- ).execute().fetchall(), [])
+ eq_(users.select().where(users.c.name == 'name3')
+ .execute().fetchall(), [])
users.update().where(users.c.name == 'name4'
).execute(name='newname')
eq_(users.select(use_labels=True).where(users.c.id
finally:
testing.db.execute('drop table speedy_users')
-
@testing.fails_on('+zxjdbc', 'psycopg2/pg8000 specific assertion')
@testing.fails_on('pypostgresql',
'psycopg2/pg8000 specific assertion')
def test_numeric_raise(self):
- stmt = text("select cast('hi' as char) as hi", typemap={'hi'
- : Numeric})
+ stmt = text(
+ "select cast('hi' as char) as hi", typemap={'hi': Numeric})
assert_raises(exc.InvalidRequestError, testing.db.execute, stmt)
- @testing.only_if("postgresql >= 8.2", "requires standard_conforming_strings")
+ @testing.only_if(
+ "postgresql >= 8.2", "requires standard_conforming_strings")
def test_serial_integer(self):
for version, type_, expected in [
else:
dialect = testing.db.dialect
- ddl_compiler = dialect.ddl_compiler(
- dialect,
- schema.CreateTable(t)
- )
+ ddl_compiler = dialect.ddl_compiler(dialect, schema.CreateTable(t))
eq_(
ddl_compiler.get_column_specification(t.c.c),
"c %s NOT NULL" % expected
)
-
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
