/usr/lib/python3/dist-packages/pg.py is in python3-pygresql 1:5.0.4-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493 2494 2495 2496 2497 2498 2499 2500 2501 2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583 2584 2585 2586 2587 2588 2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600 | #! /usr/bin/python
#
# pg.py
#
# $Id: pg.py 903 2017-01-06 16:36:05Z cito $
#
"""PyGreSQL classic interface.
This pg module implements some basic database management stuff.
It includes the _pg module and builds on it, providing the higher
level wrapper class named DB with additional functionality.
This is known as the "classic" ("old style") PyGreSQL interface.
For a DB-API 2 compliant interface use the newer pgdb module.
"""
# Copyright (c) 1997-2016 by D'Arcy J.M. Cain.
#
# Contributions made by Ch. Zwerschke and others.
#
# The notification handler is based on pgnotify which is
# Copyright (c) 2001 Ng Pheng Siong. All rights reserved.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose and without fee is hereby granted,
# provided that the above copyright notice appear in all copies and that
# both that copyright notice and this permission notice appear in
# supporting documentation.
from __future__ import print_function, division
from _pg import *
__version__ = version
import select
import warnings
import weakref
from datetime import date, time, datetime, timedelta, tzinfo
from decimal import Decimal
from math import isnan, isinf
from collections import namedtuple
from keyword import iskeyword
from operator import itemgetter
from functools import partial
from re import compile as regex
from json import loads as jsondecode, dumps as jsonencode
from uuid import UUID
try:
long
except NameError: # Python >= 3.0
long = int
try:
basestring
except NameError: # Python >= 3.0
basestring = (str, bytes)
try:
from functools import lru_cache
except ImportError: # Python < 3.2
from functools import update_wrapper
try:
from _thread import RLock
except ImportError:
class RLock: # for builds without threads
def __enter__(self): pass
def __exit__(self, exctype, excinst, exctb): pass
def lru_cache(maxsize=128):
"""Simplified functools.lru_cache decorator for one argument."""
def decorator(function):
sentinel = object()
cache = {}
get = cache.get
lock = RLock()
root = []
root_full = [root, False]
root[:] = [root, root, None, None]
if maxsize == 0:
def wrapper(arg):
res = function(arg)
return res
elif maxsize is None:
def wrapper(arg):
res = get(arg, sentinel)
if res is not sentinel:
return res
res = function(arg)
cache[arg] = res
return res
else:
def wrapper(arg):
with lock:
link = get(arg)
if link is not None:
root = root_full[0]
prev, next, _arg, res = link
prev[1] = next
next[0] = prev
last = root[0]
last[1] = root[0] = link
link[0] = last
link[1] = root
return res
res = function(arg)
with lock:
root, full = root_full
if arg in cache:
pass
elif full:
oldroot = root
oldroot[2] = arg
oldroot[3] = res
root = root_full[0] = oldroot[1]
oldarg = root[2]
oldres = root[3] # keep reference
root[2] = root[3] = None
del cache[oldarg]
cache[arg] = oldroot
else:
last = root[0]
link = [last, root, arg, res]
last[1] = root[0] = cache[arg] = link
if len(cache) >= maxsize:
root_full[1] = True
return res
wrapper.__wrapped__ = function
return update_wrapper(wrapper, function)
return decorator
# Auxiliary classes and functions that are independent from a DB connection:
try:
from collections import OrderedDict
except ImportError: # Python 2.6 or 3.0
OrderedDict = dict
class AttrDict(dict):
"""Simple read-only ordered dictionary for storing attribute names."""
def __init__(self, *args, **kw):
if len(args) > 1 or kw:
raise TypeError
items = args[0] if args else []
if isinstance(items, dict):
raise TypeError
items = list(items)
self._keys = [item[0] for item in items]
dict.__init__(self, items)
self._read_only = True
error = self._read_only_error
self.clear = self.update = error
self.pop = self.setdefault = self.popitem = error
def __setitem__(self, key, value):
if self._read_only:
self._read_only_error()
dict.__setitem__(self, key, value)
def __delitem__(self, key):
if self._read_only:
self._read_only_error()
dict.__delitem__(self, key)
def __iter__(self):
return iter(self._keys)
def keys(self):
return list(self._keys)
def values(self):
return [self[key] for key in self]
def items(self):
return [(key, self[key]) for key in self]
def iterkeys(self):
return self.__iter__()
def itervalues(self):
return iter(self.values())
def iteritems(self):
return iter(self.items())
@staticmethod
def _read_only_error(*args, **kw):
raise TypeError('This object is read-only')
else:
class AttrDict(OrderedDict):
"""Simple read-only ordered dictionary for storing attribute names."""
def __init__(self, *args, **kw):
self._read_only = False
OrderedDict.__init__(self, *args, **kw)
self._read_only = True
error = self._read_only_error
self.clear = self.update = error
self.pop = self.setdefault = self.popitem = error
def __setitem__(self, key, value):
if self._read_only:
self._read_only_error()
OrderedDict.__setitem__(self, key, value)
def __delitem__(self, key):
if self._read_only:
self._read_only_error()
OrderedDict.__delitem__(self, key)
@staticmethod
def _read_only_error(*args, **kw):
raise TypeError('This object is read-only')
try:
from inspect import signature
except ImportError: # Python < 3.3
from inspect import getargspec
def get_args(func):
return getargspec(func).args
else:
def get_args(func):
return list(signature(func).parameters)
try:
from datetime import timezone
except ImportError: # Python < 3.2
class timezone(tzinfo):
"""Simple timezone implementation."""
def __init__(self, offset, name=None):
self.offset = offset
if not name:
minutes = self.offset.days * 1440 + self.offset.seconds // 60
if minutes < 0:
hours, minutes = divmod(-minutes, 60)
hours = -hours
else:
hours, minutes = divmod(minutes, 60)
name = 'UTC%+03d:%02d' % (hours, minutes)
self.name = name
def utcoffset(self, dt):
return self.offset
def tzname(self, dt):
return self.name
def dst(self, dt):
return None
timezone.utc = timezone(timedelta(0), 'UTC')
_has_timezone = False
else:
_has_timezone = True
# time zones used in Postgres timestamptz output
_timezones = dict(CET='+0100', EET='+0200', EST='-0500',
GMT='+0000', HST='-1000', MET='+0100', MST='-0700',
UCT='+0000', UTC='+0000', WET='+0000')
def _timezone_as_offset(tz):
if tz.startswith(('+', '-')):
if len(tz) < 5:
return tz + '00'
return tz.replace(':', '')
return _timezones.get(tz, '+0000')
def _get_timezone(tz):
tz = _timezone_as_offset(tz)
minutes = 60 * int(tz[1:3]) + int(tz[3:5])
if tz[0] == '-':
minutes = -minutes
return timezone(timedelta(minutes=minutes), tz)
def _oid_key(table):
"""Build oid key from a table name."""
return 'oid(%s)' % table
class _SimpleTypes(dict):
"""Dictionary mapping pg_type names to simple type names."""
_types = {'bool': 'bool',
'bytea': 'bytea',
'date': 'date interval time timetz timestamp timestamptz'
' abstime reltime', # these are very old
'float': 'float4 float8',
'int': 'cid int2 int4 int8 oid xid',
'hstore': 'hstore', 'json': 'json jsonb', 'uuid': 'uuid',
'num': 'numeric', 'money': 'money',
'text': 'bpchar char name text varchar'}
def __init__(self):
for typ, keys in self._types.items():
for key in keys.split():
self[key] = typ
self['_%s' % key] = '%s[]' % typ
# this could be a static method in Python > 2.6
def __missing__(self, key):
return 'text'
_simpletypes = _SimpleTypes()
def _quote_if_unqualified(param, name):
"""Quote parameter representing a qualified name.
Puts a quote_ident() call around the give parameter unless
the name contains a dot, in which case the name is ambiguous
(could be a qualified name or just a name with a dot in it)
and must be quoted manually by the caller.
"""
if isinstance(name, basestring) and '.' not in name:
return 'quote_ident(%s)' % (param,)
return param
class _ParameterList(list):
"""Helper class for building typed parameter lists."""
def add(self, value, typ=None):
"""Typecast value with known database type and build parameter list.
If this is a literal value, it will be returned as is. Otherwise, a
placeholder will be returned and the parameter list will be augmented.
"""
value = self.adapt(value, typ)
if isinstance(value, Literal):
return value
self.append(value)
return '$%d' % len(self)
class Bytea(bytes):
"""Wrapper class for marking Bytea values."""
class Hstore(dict):
"""Wrapper class for marking hstore values."""
_re_quote = regex('^[Nn][Uu][Ll][Ll]$|[ ,=>]')
@classmethod
def _quote(cls, s):
if s is None:
return 'NULL'
if not s:
return '""'
s = s.replace('"', '\\"')
if cls._re_quote.search(s):
s = '"%s"' % s
return s
def __str__(self):
q = self._quote
return ','.join('%s=>%s' % (q(k), q(v)) for k, v in self.items())
class Json:
"""Wrapper class for marking Json values."""
def __init__(self, obj):
self.obj = obj
class Literal(str):
"""Wrapper class for marking literal SQL values."""
class Adapter:
"""Class providing methods for adapting parameters to the database."""
_bool_true_values = frozenset('t true 1 y yes on'.split())
_date_literals = frozenset('current_date current_time'
' current_timestamp localtime localtimestamp'.split())
_re_array_quote = regex(r'[{},"\\\s]|^[Nn][Uu][Ll][Ll]$')
_re_record_quote = regex(r'[(,"\\]')
_re_array_escape = _re_record_escape = regex(r'(["\\])')
def __init__(self, db):
self.db = weakref.proxy(db)
@classmethod
def _adapt_bool(cls, v):
"""Adapt a boolean parameter."""
if isinstance(v, basestring):
if not v:
return None
v = v.lower() in cls._bool_true_values
return 't' if v else 'f'
@classmethod
def _adapt_date(cls, v):
"""Adapt a date parameter."""
if not v:
return None
if isinstance(v, basestring) and v.lower() in cls._date_literals:
return Literal(v)
return v
@staticmethod
def _adapt_num(v):
"""Adapt a numeric parameter."""
if not v and v != 0:
return None
return v
_adapt_int = _adapt_float = _adapt_money = _adapt_num
def _adapt_bytea(self, v):
"""Adapt a bytea parameter."""
return self.db.escape_bytea(v)
def _adapt_json(self, v):
"""Adapt a json parameter."""
if not v:
return None
if isinstance(v, basestring):
return v
return self.db.encode_json(v)
@classmethod
def _adapt_text_array(cls, v):
"""Adapt a text type array parameter."""
if isinstance(v, list):
adapt = cls._adapt_text_array
return '{%s}' % ','.join(adapt(v) for v in v)
if v is None:
return 'null'
if not v:
return '""'
v = str(v)
if cls._re_array_quote.search(v):
v = '"%s"' % cls._re_array_escape.sub(r'\\\1', v)
return v
_adapt_date_array = _adapt_text_array
@classmethod
def _adapt_bool_array(cls, v):
"""Adapt a boolean array parameter."""
if isinstance(v, list):
adapt = cls._adapt_bool_array
return '{%s}' % ','.join(adapt(v) for v in v)
if v is None:
return 'null'
if isinstance(v, basestring):
if not v:
return 'null'
v = v.lower() in cls._bool_true_values
return 't' if v else 'f'
@classmethod
def _adapt_num_array(cls, v):
"""Adapt a numeric array parameter."""
if isinstance(v, list):
adapt = cls._adapt_num_array
return '{%s}' % ','.join(adapt(v) for v in v)
if not v and v != 0:
return 'null'
return str(v)
_adapt_int_array = _adapt_float_array = _adapt_money_array = \
_adapt_num_array
def _adapt_bytea_array(self, v):
"""Adapt a bytea array parameter."""
if isinstance(v, list):
return b'{' + b','.join(
self._adapt_bytea_array(v) for v in v) + b'}'
if v is None:
return b'null'
return self.db.escape_bytea(v).replace(b'\\', b'\\\\')
def _adapt_json_array(self, v):
"""Adapt a json array parameter."""
if isinstance(v, list):
adapt = self._adapt_json_array
return '{%s}' % ','.join(adapt(v) for v in v)
if not v:
return 'null'
if not isinstance(v, basestring):
v = self.db.encode_json(v)
if self._re_array_quote.search(v):
v = '"%s"' % self._re_array_escape.sub(r'\\\1', v)
return v
def _adapt_record(self, v, typ):
"""Adapt a record parameter with given type."""
typ = self.get_attnames(typ).values()
if len(typ) != len(v):
raise TypeError('Record parameter %s has wrong size' % v)
adapt = self.adapt
value = []
for v, t in zip(v, typ):
v = adapt(v, t)
if v is None:
v = ''
elif not v:
v = '""'
else:
if isinstance(v, bytes):
if str is not bytes:
v = v.decode('ascii')
else:
v = str(v)
if self._re_record_quote.search(v):
v = '"%s"' % self._re_record_escape.sub(r'\\\1', v)
value.append(v)
return '(%s)' % ','.join(value)
def adapt(self, value, typ=None):
"""Adapt a value with known database type."""
if value is not None and not isinstance(value, Literal):
if typ:
simple = self.get_simple_name(typ)
else:
typ = simple = self.guess_simple_type(value) or 'text'
try:
value = value.__pg_str__(typ)
except AttributeError:
pass
if simple == 'text':
pass
elif simple == 'record':
if isinstance(value, tuple):
value = self._adapt_record(value, typ)
elif simple.endswith('[]'):
if isinstance(value, list):
adapt = getattr(self, '_adapt_%s_array' % simple[:-2])
value = adapt(value)
else:
adapt = getattr(self, '_adapt_%s' % simple)
value = adapt(value)
return value
@staticmethod
def simple_type(name):
"""Create a simple database type with given attribute names."""
typ = DbType(name)
typ.simple = name
return typ
@staticmethod
def get_simple_name(typ):
"""Get the simple name of a database type."""
if isinstance(typ, DbType):
return typ.simple
return _simpletypes[typ]
@staticmethod
def get_attnames(typ):
"""Get the attribute names of a composite database type."""
if isinstance(typ, DbType):
return typ.attnames
return {}
@classmethod
def guess_simple_type(cls, value):
"""Try to guess which database type the given value has."""
if isinstance(value, Bytea):
return 'bytea'
if isinstance(value, basestring):
return 'text'
if isinstance(value, bool):
return 'bool'
if isinstance(value, (int, long)):
return 'int'
if isinstance(value, float):
return 'float'
if isinstance(value, Decimal):
return 'num'
if isinstance(value, (date, time, datetime, timedelta)):
return 'date'
if isinstance(value, list):
return '%s[]' % (cls.guess_simple_base_type(value) or 'text',)
if isinstance(value, tuple):
simple_type = cls.simple_type
typ = simple_type('record')
guess = cls.guess_simple_type
def get_attnames(self):
return AttrDict((str(n + 1), simple_type(guess(v)))
for n, v in enumerate(value))
typ._get_attnames = get_attnames
return typ
@classmethod
def guess_simple_base_type(cls, value):
"""Try to guess the base type of a given array."""
for v in value:
if isinstance(v, list):
typ = cls.guess_simple_base_type(v)
else:
typ = cls.guess_simple_type(v)
if typ:
return typ
def adapt_inline(self, value, nested=False):
"""Adapt a value that is put into the SQL and needs to be quoted."""
if value is None:
return 'NULL'
if isinstance(value, Literal):
return value
if isinstance(value, Bytea):
value = self.db.escape_bytea(value)
if bytes is not str: # Python >= 3.0
value = value.decode('ascii')
elif isinstance(value, Json):
if value.encode:
return value.encode()
value = self.db.encode_json(value)
elif isinstance(value, (datetime, date, time, timedelta)):
value = str(value)
if isinstance(value, basestring):
value = self.db.escape_string(value)
return "'%s'" % value
if isinstance(value, bool):
return 'true' if value else 'false'
if isinstance(value, float):
if isinf(value):
return "'-Infinity'" if value < 0 else "'Infinity'"
if isnan(value):
return "'NaN'"
return value
if isinstance(value, (int, long, Decimal)):
return value
if isinstance(value, list):
q = self.adapt_inline
s = '[%s]' if nested else 'ARRAY[%s]'
return s % ','.join(str(q(v, nested=True)) for v in value)
if isinstance(value, tuple):
q = self.adapt_inline
return '(%s)' % ','.join(str(q(v)) for v in value)
try:
value = value.__pg_repr__()
except AttributeError:
raise InterfaceError(
'Do not know how to adapt type %s' % type(value))
if isinstance(value, (tuple, list)):
value = self.adapt_inline(value)
return value
def parameter_list(self):
"""Return a parameter list for parameters with known database types.
The list has an add(value, typ) method that will build up the
list and return either the literal value or a placeholder.
"""
params = _ParameterList()
params.adapt = self.adapt
return params
def format_query(self, command, values=None, types=None, inline=False):
"""Format a database query using the given values and types."""
if not values:
return command, []
if inline and types:
raise ValueError('Typed parameters must be sent separately')
params = self.parameter_list()
if isinstance(values, (list, tuple)):
if inline:
adapt = self.adapt_inline
literals = [adapt(value) for value in values]
else:
add = params.add
literals = []
append = literals.append
if types:
if (not isinstance(types, (list, tuple)) or
len(types) != len(values)):
raise TypeError('The values and types do not match')
for value, typ in zip(values, types):
append(add(value, typ))
else:
for value in values:
append(add(value))
command %= tuple(literals)
elif isinstance(values, dict):
# we want to allow extra keys in the dictionary,
# so we first must find the values actually used in the command
used_values = {}
literals = dict.fromkeys(values, '')
for key in values:
del literals[key]
try:
command % literals
except KeyError:
used_values[key] = values[key]
literals[key] = ''
values = used_values
if inline:
adapt = self.adapt_inline
literals = dict((key, adapt(value))
for key, value in values.items())
else:
add = params.add
literals = {}
if types:
if not isinstance(types, dict):
raise TypeError('The values and types do not match')
for key in sorted(values):
literals[key] = add(values[key], types.get(key))
else:
for key in sorted(values):
literals[key] = add(values[key])
command %= literals
else:
raise TypeError('The values must be passed as tuple, list or dict')
return command, params
def cast_bool(value):
"""Cast a boolean value."""
if not get_bool():
return value
return value[0] == 't'
def cast_json(value):
"""Cast a JSON value."""
cast = get_jsondecode()
if not cast:
return value
return cast(value)
def cast_num(value):
"""Cast a numeric value."""
return (get_decimal() or float)(value)
def cast_money(value):
"""Cast a money value."""
point = get_decimal_point()
if not point:
return value
if point != '.':
value = value.replace(point, '.')
value = value.replace('(', '-')
value = ''.join(c for c in value if c.isdigit() or c in '.-')
return (get_decimal() or float)(value)
def cast_int2vector(value):
"""Cast an int2vector value."""
return [int(v) for v in value.split()]
def cast_date(value, connection):
"""Cast a date value."""
# The output format depends on the server setting DateStyle. The default
# setting ISO and the setting for German are actually unambiguous. The
# order of days and months in the other two settings is however ambiguous,
# so at least here we need to consult the setting to properly parse values.
if value == '-infinity':
return date.min
if value == 'infinity':
return date.max
value = value.split()
if value[-1] == 'BC':
return date.min
value = value[0]
if len(value) > 10:
return date.max
fmt = connection.date_format()
return datetime.strptime(value, fmt).date()
def cast_time(value):
"""Cast a time value."""
fmt = '%H:%M:%S.%f' if len(value) > 8 else '%H:%M:%S'
return datetime.strptime(value, fmt).time()
_re_timezone = regex('(.*)([+-].*)')
def cast_timetz(value):
"""Cast a timetz value."""
tz = _re_timezone.match(value)
if tz:
value, tz = tz.groups()
else:
tz = '+0000'
fmt = '%H:%M:%S.%f' if len(value) > 8 else '%H:%M:%S'
if _has_timezone:
value += _timezone_as_offset(tz)
fmt += '%z'
return datetime.strptime(value, fmt).timetz()
return datetime.strptime(value, fmt).timetz().replace(
tzinfo=_get_timezone(tz))
def cast_timestamp(value, connection):
"""Cast a timestamp value."""
if value == '-infinity':
return datetime.min
if value == 'infinity':
return datetime.max
value = value.split()
if value[-1] == 'BC':
return datetime.min
fmt = connection.date_format()
if fmt.endswith('-%Y') and len(value) > 2:
value = value[1:5]
if len(value[3]) > 4:
return datetime.max
fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
'%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
else:
if len(value[0]) > 10:
return datetime.max
fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
return datetime.strptime(' '.join(value), ' '.join(fmt))
def cast_timestamptz(value, connection):
"""Cast a timestamptz value."""
if value == '-infinity':
return datetime.min
if value == 'infinity':
return datetime.max
value = value.split()
if value[-1] == 'BC':
return datetime.min
fmt = connection.date_format()
if fmt.endswith('-%Y') and len(value) > 2:
value = value[1:]
if len(value[3]) > 4:
return datetime.max
fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
'%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
value, tz = value[:-1], value[-1]
else:
if fmt.startswith('%Y-'):
tz = _re_timezone.match(value[1])
if tz:
value[1], tz = tz.groups()
else:
tz = '+0000'
else:
value, tz = value[:-1], value[-1]
if len(value[0]) > 10:
return datetime.max
fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
if _has_timezone:
value.append(_timezone_as_offset(tz))
fmt.append('%z')
return datetime.strptime(' '.join(value), ' '.join(fmt))
return datetime.strptime(' '.join(value), ' '.join(fmt)).replace(
tzinfo=_get_timezone(tz))
_re_interval_sql_standard = regex(
'(?:([+-])?([0-9]+)-([0-9]+) ?)?'
'(?:([+-]?[0-9]+)(?!:) ?)?'
'(?:([+-])?([0-9]+):([0-9]+):([0-9]+)(?:\\.([0-9]+))?)?')
_re_interval_postgres = regex(
'(?:([+-]?[0-9]+) ?years? ?)?'
'(?:([+-]?[0-9]+) ?mons? ?)?'
'(?:([+-]?[0-9]+) ?days? ?)?'
'(?:([+-])?([0-9]+):([0-9]+):([0-9]+)(?:\\.([0-9]+))?)?')
_re_interval_postgres_verbose = regex(
'@ ?(?:([+-]?[0-9]+) ?years? ?)?'
'(?:([+-]?[0-9]+) ?mons? ?)?'
'(?:([+-]?[0-9]+) ?days? ?)?'
'(?:([+-]?[0-9]+) ?hours? ?)?'
'(?:([+-]?[0-9]+) ?mins? ?)?'
'(?:([+-])?([0-9]+)(?:\\.([0-9]+))? ?secs?)? ?(ago)?')
_re_interval_iso_8601 = regex(
'P(?:([+-]?[0-9]+)Y)?'
'(?:([+-]?[0-9]+)M)?'
'(?:([+-]?[0-9]+)D)?'
'(?:T(?:([+-]?[0-9]+)H)?'
'(?:([+-]?[0-9]+)M)?'
'(?:([+-])?([0-9]+)(?:\\.([0-9]+))?S)?)?')
def cast_interval(value):
"""Cast an interval value."""
# The output format depends on the server setting IntervalStyle, but it's
# not necessary to consult this setting to parse it. It's faster to just
# check all possible formats, and there is no ambiguity here.
m = _re_interval_iso_8601.match(value)
if m:
m = [d or '0' for d in m.groups()]
secs_ago = m.pop(5) == '-'
m = [int(d) for d in m]
years, mons, days, hours, mins, secs, usecs = m
if secs_ago:
secs = -secs
usecs = -usecs
else:
m = _re_interval_postgres_verbose.match(value)
if m:
m, ago = [d or '0' for d in m.groups()[:8]], m.group(9)
secs_ago = m.pop(5) == '-'
m = [-int(d) for d in m] if ago else [int(d) for d in m]
years, mons, days, hours, mins, secs, usecs = m
if secs_ago:
secs = - secs
usecs = -usecs
else:
m = _re_interval_postgres.match(value)
if m and any(m.groups()):
m = [d or '0' for d in m.groups()]
hours_ago = m.pop(3) == '-'
m = [int(d) for d in m]
years, mons, days, hours, mins, secs, usecs = m
if hours_ago:
hours = -hours
mins = -mins
secs = -secs
usecs = -usecs
else:
m = _re_interval_sql_standard.match(value)
if m and any(m.groups()):
m = [d or '0' for d in m.groups()]
years_ago = m.pop(0) == '-'
hours_ago = m.pop(3) == '-'
m = [int(d) for d in m]
years, mons, days, hours, mins, secs, usecs = m
if years_ago:
years = -years
mons = -mons
if hours_ago:
hours = -hours
mins = -mins
secs = -secs
usecs = -usecs
else:
raise ValueError('Cannot parse interval: %s' % value)
days += 365 * years + 30 * mons
return timedelta(days=days, hours=hours, minutes=mins,
seconds=secs, microseconds=usecs)
class Typecasts(dict):
"""Dictionary mapping database types to typecast functions.
The cast functions get passed the string representation of a value in
the database which they need to convert to a Python object. The
passed string will never be None since NULL values are already be
handled before the cast function is called.
Note that the basic types are already handled by the C extension.
They only need to be handled here as record or array components.
"""
# the default cast functions
# (str functions are ignored but have been added for faster access)
defaults = {'char': str, 'bpchar': str, 'name': str,
'text': str, 'varchar': str,
'bool': cast_bool, 'bytea': unescape_bytea,
'int2': int, 'int4': int, 'serial': int, 'int8': long, 'oid': int,
'hstore': cast_hstore, 'json': cast_json, 'jsonb': cast_json,
'float4': float, 'float8': float,
'numeric': cast_num, 'money': cast_money,
'date': cast_date, 'interval': cast_interval,
'time': cast_time, 'timetz': cast_timetz,
'timestamp': cast_timestamp, 'timestamptz': cast_timestamptz,
'int2vector': cast_int2vector, 'uuid': UUID,
'anyarray': cast_array, 'record': cast_record}
connection = None # will be set in a connection specific instance
def __missing__(self, typ):
"""Create a cast function if it is not cached.
Note that this class never raises a KeyError,
but returns None when no special cast function exists.
"""
if not isinstance(typ, str):
raise TypeError('Invalid type: %s' % typ)
cast = self.defaults.get(typ)
if cast:
# store default for faster access
cast = self._add_connection(cast)
self[typ] = cast
elif typ.startswith('_'):
base_cast = self[typ[1:]]
cast = self.create_array_cast(base_cast)
if base_cast:
self[typ] = cast
else:
attnames = self.get_attnames(typ)
if attnames:
casts = [self[v.pgtype] for v in attnames.values()]
cast = self.create_record_cast(typ, attnames, casts)
self[typ] = cast
return cast
@staticmethod
def _needs_connection(func):
"""Check if a typecast function needs a connection argument."""
try:
args = get_args(func)
except (TypeError, ValueError):
return False
else:
return 'connection' in args[1:]
def _add_connection(self, cast):
"""Add a connection argument to the typecast function if necessary."""
if not self.connection or not self._needs_connection(cast):
return cast
return partial(cast, connection=self.connection)
def get(self, typ, default=None):
"""Get the typecast function for the given database type."""
return self[typ] or default
def set(self, typ, cast):
"""Set a typecast function for the specified database type(s)."""
if isinstance(typ, basestring):
typ = [typ]
if cast is None:
for t in typ:
self.pop(t, None)
self.pop('_%s' % t, None)
else:
if not callable(cast):
raise TypeError("Cast parameter must be callable")
for t in typ:
self[t] = self._add_connection(cast)
self.pop('_%s' % t, None)
def reset(self, typ=None):
"""Reset the typecasts for the specified type(s) to their defaults.
When no type is specified, all typecasts will be reset.
"""
if typ is None:
self.clear()
else:
if isinstance(typ, basestring):
typ = [typ]
for t in typ:
self.pop(t, None)
@classmethod
def get_default(cls, typ):
"""Get the default typecast function for the given database type."""
return cls.defaults.get(typ)
@classmethod
def set_default(cls, typ, cast):
"""Set a default typecast function for the given database type(s)."""
if isinstance(typ, basestring):
typ = [typ]
defaults = cls.defaults
if cast is None:
for t in typ:
defaults.pop(t, None)
defaults.pop('_%s' % t, None)
else:
if not callable(cast):
raise TypeError("Cast parameter must be callable")
for t in typ:
defaults[t] = cast
defaults.pop('_%s' % t, None)
def get_attnames(self, typ):
"""Return the fields for the given record type.
This method will be replaced with the get_attnames() method of DbTypes.
"""
return {}
def dateformat(self):
"""Return the current date format.
This method will be replaced with the dateformat() method of DbTypes.
"""
return '%Y-%m-%d'
def create_array_cast(self, basecast):
"""Create an array typecast for the given base cast."""
cast_array = self['anyarray']
def cast(v):
return cast_array(v, basecast)
return cast
def create_record_cast(self, name, fields, casts):
"""Create a named record typecast for the given fields and casts."""
cast_record = self['record']
record = namedtuple(name, fields)
def cast(v):
return record(*cast_record(v, casts))
return cast
def get_typecast(typ):
"""Get the global typecast function for the given database type(s)."""
return Typecasts.get_default(typ)
def set_typecast(typ, cast):
"""Set a global typecast function for the given database type(s).
Note that connections cache cast functions. To be sure a global change
is picked up by a running connection, call db.db_types.reset_typecast().
"""
Typecasts.set_default(typ, cast)
class DbType(str):
"""Class augmenting the simple type name with additional info.
The following additional information is provided:
oid: the PostgreSQL type OID
pgtype: the PostgreSQL type name
regtype: the regular type name
simple: the simple PyGreSQL type name
typtype: b = base type, c = composite type etc.
category: A = Array, b = Boolean, C = Composite etc.
delim: delimiter for array types
relid: corresponding table for composite types
attnames: attributes for composite types
"""
@property
def attnames(self):
"""Get names and types of the fields of a composite type."""
return self._get_attnames(self)
class DbTypes(dict):
"""Cache for PostgreSQL data types.
This cache maps type OIDs and names to DbType objects containing
information on the associated database type.
"""
_num_types = frozenset('int float num money'
' int2 int4 int8 float4 float8 numeric money'.split())
def __init__(self, db):
"""Initialize type cache for connection."""
super(DbTypes, self).__init__()
self._db = weakref.proxy(db)
self._regtypes = False
self._typecasts = Typecasts()
self._typecasts.get_attnames = self.get_attnames
self._typecasts.connection = self._db
if db.server_version < 80400:
# older remote databases (not officially supported)
self._query_pg_type = (
"SELECT oid, typname, typname::text::regtype,"
" typtype, null as typcategory, typdelim, typrelid"
" FROM pg_type WHERE oid=%s::regtype")
else:
self._query_pg_type = (
"SELECT oid, typname, typname::regtype,"
" typtype, typcategory, typdelim, typrelid"
" FROM pg_type WHERE oid=%s::regtype")
def add(self, oid, pgtype, regtype,
typtype, category, delim, relid):
"""Create a PostgreSQL type name with additional info."""
if oid in self:
return self[oid]
simple = 'record' if relid else _simpletypes[pgtype]
typ = DbType(regtype if self._regtypes else simple)
typ.oid = oid
typ.simple = simple
typ.pgtype = pgtype
typ.regtype = regtype
typ.typtype = typtype
typ.category = category
typ.delim = delim
typ.relid = relid
typ._get_attnames = self.get_attnames
return typ
def __missing__(self, key):
"""Get the type info from the database if it is not cached."""
try:
q = self._query_pg_type % (_quote_if_unqualified('$1', key),)
res = self._db.query(q, (key,)).getresult()
except ProgrammingError:
res = None
if not res:
raise KeyError('Type %s could not be found' % key)
res = res[0]
typ = self.add(*res)
self[typ.oid] = self[typ.pgtype] = typ
return typ
def get(self, key, default=None):
"""Get the type even if it is not cached."""
try:
return self[key]
except KeyError:
return default
def get_attnames(self, typ):
"""Get names and types of the fields of a composite type."""
if not isinstance(typ, DbType):
typ = self.get(typ)
if not typ:
return None
if not typ.relid:
return None
return self._db.get_attnames(typ.relid, with_oid=False)
def get_typecast(self, typ):
"""Get the typecast function for the given database type."""
return self._typecasts.get(typ)
def set_typecast(self, typ, cast):
"""Set a typecast function for the specified database type(s)."""
self._typecasts.set(typ, cast)
def reset_typecast(self, typ=None):
"""Reset the typecast function for the specified database type(s)."""
self._typecasts.reset(typ)
def typecast(self, value, typ):
"""Cast the given value according to the given database type."""
if value is None:
# for NULL values, no typecast is necessary
return None
if not isinstance(typ, DbType):
typ = self.get(typ)
if typ:
typ = typ.pgtype
cast = self.get_typecast(typ) if typ else None
if not cast or cast is str:
# no typecast is necessary
return value
return cast(value)
_re_fieldname = regex('^[A-Za-z][_a-zA-Z0-9]*$')
# The result rows for database operations are returned as named tuples
# by default. Since creating namedtuple classes is a somewhat expensive
# operation, we cache up to 1024 of these classes by default.
@lru_cache(maxsize=1024)
def _row_factory(names):
"""Get a namedtuple factory for row results with the given names."""
try:
try:
return namedtuple('Row', names, rename=True)._make
except TypeError: # Python 2.6 and 3.0 do not support rename
names = [v if _re_fieldname.match(v) and not iskeyword(v)
else 'column_%d' % (n,)
for n, v in enumerate(names)]
return namedtuple('Row', names)._make
except ValueError: # there is still a problem with the field names
names = ['column_%d' % (n,) for n in range(len(names))]
return namedtuple('Row', names)._make
def set_row_factory_size(maxsize):
"""Change the size of the namedtuple factory cache.
If maxsize is set to None, the cache can grow without bound.
"""
global _row_factory
_row_factory = lru_cache(maxsize)(_row_factory.__wrapped__)
def _namedresult(q):
"""Get query result as named tuples."""
row = _row_factory(q.listfields())
return [row(r) for r in q.getresult()]
class _MemoryQuery:
"""Class that embodies a given query result."""
def __init__(self, result, fields):
"""Create query from given result rows and field names."""
self.result = result
self.fields = tuple(fields)
def listfields(self):
"""Return the stored field names of this query."""
return self.fields
def getresult(self):
"""Return the stored result of this query."""
return self.result
def _db_error(msg, cls=DatabaseError):
"""Return DatabaseError with empty sqlstate attribute."""
error = cls(msg)
error.sqlstate = None
return error
def _int_error(msg):
"""Return InternalError."""
return _db_error(msg, InternalError)
def _prg_error(msg):
"""Return ProgrammingError."""
return _db_error(msg, ProgrammingError)
# Initialize the C module
set_namedresult(_namedresult)
set_decimal(Decimal)
set_jsondecode(jsondecode)
# The notification handler
class NotificationHandler(object):
"""A PostgreSQL client-side asynchronous notification handler."""
def __init__(self, db, event, callback=None,
arg_dict=None, timeout=None, stop_event=None):
"""Initialize the notification handler.
You must pass a PyGreSQL database connection, the name of an
event (notification channel) to listen for and a callback function.
You can also specify a dictionary arg_dict that will be passed as
the single argument to the callback function, and a timeout value
in seconds (a floating point number denotes fractions of seconds).
If it is absent or None, the callers will never time out. If the
timeout is reached, the callback function will be called with a
single argument that is None. If you set the timeout to zero,
the handler will poll notifications synchronously and return.
You can specify the name of the event that will be used to signal
the handler to stop listening as stop_event. By default, it will
be the event name prefixed with 'stop_'.
"""
self.db = db
self.event = event
self.stop_event = stop_event or 'stop_%s' % event
self.listening = False
self.callback = callback
if arg_dict is None:
arg_dict = {}
self.arg_dict = arg_dict
self.timeout = timeout
def __del__(self):
self.unlisten()
def close(self):
"""Stop listening and close the connection."""
if self.db:
self.unlisten()
self.db.close()
self.db = None
def listen(self):
"""Start listening for the event and the stop event."""
if not self.listening:
self.db.query('listen "%s"' % self.event)
self.db.query('listen "%s"' % self.stop_event)
self.listening = True
def unlisten(self):
"""Stop listening for the event and the stop event."""
if self.listening:
self.db.query('unlisten "%s"' % self.event)
self.db.query('unlisten "%s"' % self.stop_event)
self.listening = False
def notify(self, db=None, stop=False, payload=None):
"""Generate a notification.
Optionally, you can pass a payload with the notification.
If you set the stop flag, a stop notification will be sent that
will cause the handler to stop listening.
Note: If the notification handler is running in another thread, you
must pass a different database connection since PyGreSQL database
connections are not thread-safe.
"""
if self.listening:
if not db:
db = self.db
q = 'notify "%s"' % (self.stop_event if stop else self.event)
if payload:
q += ", '%s'" % payload
return db.query(q)
def __call__(self):
"""Invoke the notification handler.
The handler is a loop that listens for notifications on the event
and stop event channels. When either of these notifications are
received, its associated 'pid', 'event' and 'extra' (the payload
passed with the notification) are inserted into its arg_dict
dictionary and the callback is invoked with this dictionary as
a single argument. When the handler receives a stop event, it
stops listening to both events and return.
In the special case that the timeout of the handler has been set
to zero, the handler will poll all events synchronously and return.
If will keep listening until it receives a stop event.
Note: If you run this loop in another thread, don't use the same
database connection for database operations in the main thread.
"""
self.listen()
poll = self.timeout == 0
if not poll:
rlist = [self.db.fileno()]
while self.listening:
if poll or select.select(rlist, [], [], self.timeout)[0]:
while self.listening:
notice = self.db.getnotify()
if not notice: # no more messages
break
event, pid, extra = notice
if event not in (self.event, self.stop_event):
self.unlisten()
raise _db_error(
'Listening for "%s" and "%s", but notified of "%s"'
% (self.event, self.stop_event, event))
if event == self.stop_event:
self.unlisten()
self.arg_dict.update(pid=pid, event=event, extra=extra)
self.callback(self.arg_dict)
if poll:
break
else: # we timed out
self.unlisten()
self.callback(None)
def pgnotify(*args, **kw):
"""Same as NotificationHandler, under the traditional name."""
warnings.warn("pgnotify is deprecated, use NotificationHandler instead",
DeprecationWarning, stacklevel=2)
return NotificationHandler(*args, **kw)
# The actual PostGreSQL database connection interface:
class DB:
"""Wrapper class for the _pg connection type."""
db = None # invalid fallback for underlying connection
def __init__(self, *args, **kw):
"""Create a new connection
You can pass either the connection parameters or an existing
_pg or pgdb connection. This allows you to use the methods
of the classic pg interface with a DB-API 2 pgdb connection.
"""
if not args and len(kw) == 1:
db = kw.get('db')
elif not kw and len(args) == 1:
db = args[0]
else:
db = None
if db:
if isinstance(db, DB):
db = db.db
else:
try:
db = db._cnx
except AttributeError:
pass
if not db or not hasattr(db, 'db') or not hasattr(db, 'query'):
db = connect(*args, **kw)
self._closeable = True
else:
self._closeable = False
self.db = db
self.dbname = db.db
self._regtypes = False
self._attnames = {}
self._pkeys = {}
self._privileges = {}
self._args = args, kw
self.adapter = Adapter(self)
self.dbtypes = DbTypes(self)
if db.server_version < 80400:
# support older remote data bases
self._query_attnames = (
"SELECT a.attname, t.oid, t.typname, t.typname::text::regtype,"
" t.typtype, null as typcategory, t.typdelim, t.typrelid"
" FROM pg_attribute a"
" JOIN pg_type t ON t.oid = a.atttypid"
" WHERE a.attrelid = %s::regclass AND %s"
" AND NOT a.attisdropped ORDER BY a.attnum")
else:
self._query_attnames = (
"SELECT a.attname, t.oid, t.typname, t.typname::regtype,"
" t.typtype, t.typcategory, t.typdelim, t.typrelid"
" FROM pg_attribute a"
" JOIN pg_type t ON t.oid = a.atttypid"
" WHERE a.attrelid = %s::regclass AND %s"
" AND NOT a.attisdropped ORDER BY a.attnum")
db.set_cast_hook(self.dbtypes.typecast)
self.debug = None # For debugging scripts, this can be set
# * to a string format specification (e.g. in CGI set to "%s<BR>"),
# * to a file object to write debug statements or
# * to a callable object which takes a string argument
# * to any other true value to just print debug statements
def __getattr__(self, name):
# All undefined members are same as in underlying connection:
if self.db:
return getattr(self.db, name)
else:
raise _int_error('Connection is not valid')
def __dir__(self):
# Custom dir function including the attributes of the connection:
attrs = set(self.__class__.__dict__)
attrs.update(self.__dict__)
attrs.update(dir(self.db))
return sorted(attrs)
# Context manager methods
def __enter__(self):
"""Enter the runtime context. This will start a transaction."""
self.begin()
return self
def __exit__(self, et, ev, tb):
"""Exit the runtime context. This will end the transaction."""
if et is None and ev is None and tb is None:
self.commit()
else:
self.rollback()
def __del__(self):
try:
db = self.db
except AttributeError:
db = None
if db:
db.set_cast_hook(None)
if self._closeable:
db.close()
# Auxiliary methods
def _do_debug(self, *args):
"""Print a debug message"""
if self.debug:
s = '\n'.join(str(arg) for arg in args)
if isinstance(self.debug, basestring):
print(self.debug % s)
elif hasattr(self.debug, 'write'):
self.debug.write(s + '\n')
elif callable(self.debug):
self.debug(s)
else:
print(s)
def _escape_qualified_name(self, s):
"""Escape a qualified name.
Escapes the name for use as an SQL identifier, unless the
name contains a dot, in which case the name is ambiguous
(could be a qualified name or just a name with a dot in it)
and must be quoted manually by the caller.
"""
if '.' not in s:
s = self.escape_identifier(s)
return s
@staticmethod
def _make_bool(d):
"""Get boolean value corresponding to d."""
return bool(d) if get_bool() else ('t' if d else 'f')
def _list_params(self, params):
"""Create a human readable parameter list."""
return ', '.join('$%d=%r' % (n, v) for n, v in enumerate(params, 1))
# Public methods
# escape_string and escape_bytea exist as methods,
# so we define unescape_bytea as a method as well
unescape_bytea = staticmethod(unescape_bytea)
def decode_json(self, s):
"""Decode a JSON string coming from the database."""
return (get_jsondecode() or jsondecode)(s)
def encode_json(self, d):
"""Encode a JSON string for use within SQL."""
return jsonencode(d)
def close(self):
"""Close the database connection."""
# Wraps shared library function so we can track state.
if self._closeable:
if self.db:
self.db.set_cast_hook(None)
self.db.close()
self.db = None
else:
raise _int_error('Connection already closed')
def reset(self):
"""Reset connection with current parameters.
All derived queries and large objects derived from this connection
will not be usable after this call.
"""
if self.db:
self.db.reset()
else:
raise _int_error('Connection already closed')
def reopen(self):
"""Reopen connection to the database.
Used in case we need another connection to the same database.
Note that we can still reopen a database that we have closed.
"""
# There is no such shared library function.
if self._closeable:
db = connect(*self._args[0], **self._args[1])
if self.db:
self.db.set_cast_hook(None)
self.db.close()
self.db = db
def begin(self, mode=None):
"""Begin a transaction."""
qstr = 'BEGIN'
if mode:
qstr += ' ' + mode
return self.query(qstr)
start = begin
def commit(self):
"""Commit the current transaction."""
return self.query('COMMIT')
end = commit
def rollback(self, name=None):
"""Roll back the current transaction."""
qstr = 'ROLLBACK'
if name:
qstr += ' TO ' + name
return self.query(qstr)
abort = rollback
def savepoint(self, name):
"""Define a new savepoint within the current transaction."""
return self.query('SAVEPOINT ' + name)
def release(self, name):
"""Destroy a previously defined savepoint."""
return self.query('RELEASE ' + name)
def get_parameter(self, parameter):
"""Get the value of a run-time parameter.
If the parameter is a string, the return value will also be a string
that is the current setting of the run-time parameter with that name.
You can get several parameters at once by passing a list, set or dict.
When passing a list of parameter names, the return value will be a
corresponding list of parameter settings. When passing a set of
parameter names, a new dict will be returned, mapping these parameter
names to their settings. Finally, if you pass a dict as parameter,
its values will be set to the current parameter settings corresponding
to its keys.
By passing the special name 'all' as the parameter, you can get a dict
of all existing configuration parameters.
"""
if isinstance(parameter, basestring):
parameter = [parameter]
values = None
elif isinstance(parameter, (list, tuple)):
values = []
elif isinstance(parameter, (set, frozenset)):
values = {}
elif isinstance(parameter, dict):
values = parameter
else:
raise TypeError(
'The parameter must be a string, list, set or dict')
if not parameter:
raise TypeError('No parameter has been specified')
params = {} if isinstance(values, dict) else []
for key in parameter:
param = key.strip().lower() if isinstance(
key, basestring) else None
if not param:
raise TypeError('Invalid parameter')
if param == 'all':
q = 'SHOW ALL'
values = self.db.query(q).getresult()
values = dict(value[:2] for value in values)
break
if isinstance(values, dict):
params[param] = key
else:
params.append(param)
else:
for param in params:
q = 'SHOW %s' % (param,)
value = self.db.query(q).getresult()[0][0]
if values is None:
values = value
elif isinstance(values, list):
values.append(value)
else:
values[params[param]] = value
return values
def set_parameter(self, parameter, value=None, local=False):
"""Set the value of a run-time parameter.
If the parameter and the value are strings, the run-time parameter
will be set to that value. If no value or None is passed as a value,
then the run-time parameter will be restored to its default value.
You can set several parameters at once by passing a list of parameter
names, together with a single value that all parameters should be
set to or with a corresponding list of values. You can also pass
the parameters as a set if you only provide a single value.
Finally, you can pass a dict with parameter names as keys. In this
case, you should not pass a value, since the values for the parameters
will be taken from the dict.
By passing the special name 'all' as the parameter, you can reset
all existing settable run-time parameters to their default values.
If you set local to True, then the command takes effect for only the
current transaction. After commit() or rollback(), the session-level
setting takes effect again. Setting local to True will appear to
have no effect if it is executed outside a transaction, since the
transaction will end immediately.
"""
if isinstance(parameter, basestring):
parameter = {parameter: value}
elif isinstance(parameter, (list, tuple)):
if isinstance(value, (list, tuple)):
parameter = dict(zip(parameter, value))
else:
parameter = dict.fromkeys(parameter, value)
elif isinstance(parameter, (set, frozenset)):
if isinstance(value, (list, tuple, set, frozenset)):
value = set(value)
if len(value) == 1:
value = value.pop()
if not(value is None or isinstance(value, basestring)):
raise ValueError('A single value must be specified'
' when parameter is a set')
parameter = dict.fromkeys(parameter, value)
elif isinstance(parameter, dict):
if value is not None:
raise ValueError('A value must not be specified'
' when parameter is a dictionary')
else:
raise TypeError(
'The parameter must be a string, list, set or dict')
if not parameter:
raise TypeError('No parameter has been specified')
params = {}
for key, value in parameter.items():
param = key.strip().lower() if isinstance(
key, basestring) else None
if not param:
raise TypeError('Invalid parameter')
if param == 'all':
if value is not None:
raise ValueError('A value must ot be specified'
" when parameter is 'all'")
params = {'all': None}
break
params[param] = value
local = ' LOCAL' if local else ''
for param, value in params.items():
if value is None:
q = 'RESET%s %s' % (local, param)
else:
q = 'SET%s %s TO %s' % (local, param, value)
self._do_debug(q)
self.db.query(q)
def query(self, command, *args):
"""Execute a SQL command string.
This method simply sends a SQL query to the database. If the query is
an insert statement that inserted exactly one row into a table that
has OIDs, the return value is the OID of the newly inserted row.
If the query is an update or delete statement, or an insert statement
that did not insert exactly one row in a table with OIDs, then the
number of rows affected is returned as a string. If it is a statement
that returns rows as a result (usually a select statement, but maybe
also an "insert/update ... returning" statement), this method returns
a Query object that can be accessed via getresult() or dictresult()
or simply printed. Otherwise, it returns `None`.
The query can contain numbered parameters of the form $1 in place
of any data constant. Arguments given after the query string will
be substituted for the corresponding numbered parameter. Parameter
values can also be given as a single list or tuple argument.
"""
# Wraps shared library function for debugging.
if not self.db:
raise _int_error('Connection is not valid')
if args:
self._do_debug(command, args)
return self.db.query(command, args)
self._do_debug(command)
return self.db.query(command)
def query_formatted(self, command,
parameters=None, types=None, inline=False):
"""Execute a formatted SQL command string.
Similar to query, but using Python format placeholders of the form
%s or %(names)s instead of PostgreSQL placeholders of the form $1.
The parameters must be passed as a tuple, list or dict. You can
also pass a corresponding tuple, list or dict of database types in
order to format the parameters properly in case there is ambiguity.
If you set inline to True, the parameters will be sent to the database
embedded in the SQL command, otherwise they will be sent separately.
"""
return self.query(*self.adapter.format_query(
command, parameters, types, inline))
def pkey(self, table, composite=False, flush=False):
"""Get or set the primary key of a table.
Single primary keys are returned as strings unless you
set the composite flag. Composite primary keys are always
represented as tuples. Note that this raises a KeyError
if the table does not have a primary key.
If flush is set then the internal cache for primary keys will
be flushed. This may be necessary after the database schema or
the search path has been changed.
"""
pkeys = self._pkeys
if flush:
pkeys.clear()
self._do_debug('The pkey cache has been flushed')
try: # cache lookup
pkey = pkeys[table]
except KeyError: # cache miss, check the database
q = ("SELECT a.attname, a.attnum, i.indkey FROM pg_index i"
" JOIN pg_attribute a ON a.attrelid = i.indrelid"
" AND a.attnum = ANY(i.indkey)"
" AND NOT a.attisdropped"
" WHERE i.indrelid=%s::regclass"
" AND i.indisprimary ORDER BY a.attnum") % (
_quote_if_unqualified('$1', table),)
pkey = self.db.query(q, (table,)).getresult()
if not pkey:
raise KeyError('Table %s has no primary key' % table)
# we want to use the order defined in the primary key index here,
# not the order as defined by the columns in the table
if len(pkey) > 1:
indkey = pkey[0][2]
pkey = sorted(pkey, key=lambda row: indkey.index(row[1]))
pkey = tuple(row[0] for row in pkey)
else:
pkey = pkey[0][0]
pkeys[table] = pkey # cache it
if composite and not isinstance(pkey, tuple):
pkey = (pkey,)
return pkey
def get_databases(self):
"""Get list of databases in the system."""
return [s[0] for s in
self.db.query('SELECT datname FROM pg_database').getresult()]
def get_relations(self, kinds=None, system=False):
"""Get list of relations in connected database of specified kinds.
If kinds is None or empty, all kinds of relations are returned.
Otherwise kinds can be a string or sequence of type letters
specifying which kind of relations you want to list.
Set the system flag if you want to get the system relations as well.
"""
where = []
if kinds:
where.append("r.relkind IN (%s)" %
','.join("'%s'" % k for k in kinds))
if not system:
where.append("s.nspname NOT SIMILAR"
" TO 'pg/_%|information/_schema' ESCAPE '/'")
where = " WHERE %s" % ' AND '.join(where) if where else ''
q = ("SELECT quote_ident(s.nspname)||'.'||quote_ident(r.relname)"
" FROM pg_class r"
" JOIN pg_namespace s ON s.oid = r.relnamespace%s"
" ORDER BY s.nspname, r.relname") % where
return [r[0] for r in self.db.query(q).getresult()]
def get_tables(self, system=False):
"""Return list of tables in connected database.
Set the system flag if you want to get the system tables as well.
"""
return self.get_relations('r', system)
def get_attnames(self, table, with_oid=True, flush=False):
"""Given the name of a table, dig out the set of attribute names.
Returns a read-only dictionary of attribute names (the names are
the keys, the values are the names of the attributes' types)
with the column names in the proper order if you iterate over it.
If flush is set, then the internal cache for attribute names will
be flushed. This may be necessary after the database schema or
the search path has been changed.
By default, only a limited number of simple types will be returned.
You can get the regular types after calling use_regtypes(True).
"""
attnames = self._attnames
if flush:
attnames.clear()
self._do_debug('The attnames cache has been flushed')
try: # cache lookup
names = attnames[table]
except KeyError: # cache miss, check the database
q = "a.attnum > 0"
if with_oid:
q = "(%s OR a.attname = 'oid')" % q
q = self._query_attnames % (_quote_if_unqualified('$1', table), q)
names = self.db.query(q, (table,)).getresult()
types = self.dbtypes
names = ((name[0], types.add(*name[1:])) for name in names)
names = AttrDict(names)
attnames[table] = names # cache it
return names
def use_regtypes(self, regtypes=None):
"""Use regular type names instead of simplified type names."""
if regtypes is None:
return self.dbtypes._regtypes
else:
regtypes = bool(regtypes)
if regtypes != self.dbtypes._regtypes:
self.dbtypes._regtypes = regtypes
self._attnames.clear()
self.dbtypes.clear()
return regtypes
def has_table_privilege(self, table, privilege='select', flush=False):
"""Check whether current user has specified table privilege.
If flush is set, then the internal cache for table privileges will
be flushed. This may be necessary after privileges have been changed.
"""
privileges = self._privileges
if flush:
privileges.clear()
self._do_debug('The privileges cache has been flushed')
privilege = privilege.lower()
try: # ask cache
ret = privileges[table, privilege]
except KeyError: # cache miss, ask the database
q = "SELECT has_table_privilege(%s, $2)" % (
_quote_if_unqualified('$1', table),)
q = self.db.query(q, (table, privilege))
ret = q.getresult()[0][0] == self._make_bool(True)
privileges[table, privilege] = ret # cache it
return ret
def get(self, table, row, keyname=None):
"""Get a row from a database table or view.
This method is the basic mechanism to get a single row. It assumes
that the keyname specifies a unique row. It must be the name of a
single column or a tuple of column names. If the keyname is not
specified, then the primary key for the table is used.
If row is a dictionary, then the value for the key is taken from it.
Otherwise, the row must be a single value or a tuple of values
corresponding to the passed keyname or primary key. The fetched row
from the table will be returned as a new dictionary or used to replace
the existing values when row was passed as a dictionary.
The OID is also put into the dictionary if the table has one, but
in order to allow the caller to work with multiple tables, it is
munged as "oid(table)" using the actual name of the table.
"""
if table.endswith('*'): # hint for descendant tables can be ignored
table = table[:-1].rstrip()
attnames = self.get_attnames(table)
qoid = _oid_key(table) if 'oid' in attnames else None
if keyname and isinstance(keyname, basestring):
keyname = (keyname,)
if qoid and isinstance(row, dict) and qoid in row and 'oid' not in row:
row['oid'] = row[qoid]
if not keyname:
try: # if keyname is not specified, try using the primary key
keyname = self.pkey(table, True)
except KeyError: # the table has no primary key
# try using the oid instead
if qoid and isinstance(row, dict) and 'oid' in row:
keyname = ('oid',)
else:
raise _prg_error('Table %s has no primary key' % table)
else: # the table has a primary key
# check whether all key columns have values
if isinstance(row, dict) and not set(keyname).issubset(row):
# try using the oid instead
if qoid and 'oid' in row:
keyname = ('oid',)
else:
raise KeyError(
'Missing value in row for specified keyname')
if not isinstance(row, dict):
if not isinstance(row, (tuple, list)):
row = [row]
if len(keyname) != len(row):
raise KeyError(
'Differing number of items in keyname and row')
row = dict(zip(keyname, row))
params = self.adapter.parameter_list()
adapt = params.add
col = self.escape_identifier
what = 'oid, *' if qoid else '*'
where = ' AND '.join('%s = %s' % (
col(k), adapt(row[k], attnames[k])) for k in keyname)
if 'oid' in row:
if qoid:
row[qoid] = row['oid']
del row['oid']
q = 'SELECT %s FROM %s WHERE %s LIMIT 1' % (
what, self._escape_qualified_name(table), where)
self._do_debug(q, params)
q = self.db.query(q, params)
res = q.dictresult()
if not res:
raise _db_error('No such record in %s\nwhere %s\nwith %s' % (
table, where, self._list_params(params)))
for n, value in res[0].items():
if qoid and n == 'oid':
n = qoid
row[n] = value
return row
def insert(self, table, row=None, **kw):
"""Insert a row into a database table.
This method inserts a row into a table. The name of the table must
be passed as the first parameter. The other parameters are used for
providing the data of the row that shall be inserted into the table.
If a dictionary is supplied as the second parameter, it starts with
that. Otherwise it uses a blank dictionary. Either way the dictionary
is updated from the keywords.
The dictionary is then reloaded with the values actually inserted in
order to pick up values modified by rules, triggers, etc.
"""
if table.endswith('*'): # hint for descendant tables can be ignored
table = table[:-1].rstrip()
if row is None:
row = {}
row.update(kw)
if 'oid' in row:
del row['oid'] # do not insert oid
attnames = self.get_attnames(table)
qoid = _oid_key(table) if 'oid' in attnames else None
params = self.adapter.parameter_list()
adapt = params.add
col = self.escape_identifier
names, values = [], []
for n in attnames:
if n in row:
names.append(col(n))
values.append(adapt(row[n], attnames[n]))
if not names:
raise _prg_error('No column found that can be inserted')
names, values = ', '.join(names), ', '.join(values)
ret = 'oid, *' if qoid else '*'
q = 'INSERT INTO %s (%s) VALUES (%s) RETURNING %s' % (
self._escape_qualified_name(table), names, values, ret)
self._do_debug(q, params)
q = self.db.query(q, params)
res = q.dictresult()
if res: # this should always be true
for n, value in res[0].items():
if qoid and n == 'oid':
n = qoid
row[n] = value
return row
def update(self, table, row=None, **kw):
"""Update an existing row in a database table.
Similar to insert, but updates an existing row. The update is based
on the primary key of the table or the OID value as munged by get()
or passed as keyword. The OID will take precedence if provided, so
that it is possible to update the primary key itself.
The dictionary is then modified to reflect any changes caused by the
update due to triggers, rules, default values, etc.
"""
if table.endswith('*'):
table = table[:-1].rstrip() # need parent table name
attnames = self.get_attnames(table)
qoid = _oid_key(table) if 'oid' in attnames else None
if row is None:
row = {}
elif 'oid' in row:
del row['oid'] # only accept oid key from named args for safety
row.update(kw)
if qoid and qoid in row and 'oid' not in row:
row['oid'] = row[qoid]
if qoid and 'oid' in row: # try using the oid
keyname = ('oid',)
else: # try using the primary key
try:
keyname = self.pkey(table, True)
except KeyError: # the table has no primary key
raise _prg_error('Table %s has no primary key' % table)
# check whether all key columns have values
if not set(keyname).issubset(row):
raise KeyError('Missing value for primary key in row')
params = self.adapter.parameter_list()
adapt = params.add
col = self.escape_identifier
where = ' AND '.join('%s = %s' % (
col(k), adapt(row[k], attnames[k])) for k in keyname)
if 'oid' in row:
if qoid:
row[qoid] = row['oid']
del row['oid']
values = []
keyname = set(keyname)
for n in attnames:
if n in row and n not in keyname:
values.append('%s = %s' % (col(n), adapt(row[n], attnames[n])))
if not values:
return row
values = ', '.join(values)
ret = 'oid, *' if qoid else '*'
q = 'UPDATE %s SET %s WHERE %s RETURNING %s' % (
self._escape_qualified_name(table), values, where, ret)
self._do_debug(q, params)
q = self.db.query(q, params)
res = q.dictresult()
if res: # may be empty when row does not exist
for n, value in res[0].items():
if qoid and n == 'oid':
n = qoid
row[n] = value
return row
def upsert(self, table, row=None, **kw):
"""Insert a row into a database table with conflict resolution
This method inserts a row into a table, but instead of raising a
ProgrammingError exception in case a row with the same primary key
already exists, an update will be executed instead. This will be
performed as a single atomic operation on the database, so race
conditions can be avoided.
Like the insert method, the first parameter is the name of the
table and the second parameter can be used to pass the values to
be inserted as a dictionary.
Unlike the insert und update statement, keyword parameters are not
used to modify the dictionary, but to specify which columns shall
be updated in case of a conflict, and in which way:
A value of False or None means the column shall not be updated,
a value of True means the column shall be updated with the value
that has been proposed for insertion, i.e. has been passed as value
in the dictionary. Columns that are not specified by keywords but
appear as keys in the dictionary are also updated like in the case
keywords had been passed with the value True.
So if in the case of a conflict you want to update every column that
has been passed in the dictionary row, you would call upsert(table, row).
If you don't want to do anything in case of a conflict, i.e. leave
the existing row as it is, call upsert(table, row, **dict.fromkeys(row)).
If you need more fine-grained control of what gets updated, you can
also pass strings in the keyword parameters. These strings will
be used as SQL expressions for the update columns. In these
expressions you can refer to the value that already exists in
the table by prefixing the column name with "included.", and to
the value that has been proposed for insertion by prefixing the
column name with the "excluded."
The dictionary is modified in any case to reflect the values in
the database after the operation has completed.
Note: The method uses the PostgreSQL "upsert" feature which is
only available since PostgreSQL 9.5.
"""
if table.endswith('*'): # hint for descendant tables can be ignored
table = table[:-1].rstrip()
if row is None:
row = {}
if 'oid' in row:
del row['oid'] # do not insert oid
if 'oid' in kw:
del kw['oid'] # do not update oid
attnames = self.get_attnames(table)
qoid = _oid_key(table) if 'oid' in attnames else None
params = self.adapter.parameter_list()
adapt = params.add
col = self.escape_identifier
names, values, updates = [], [], []
for n in attnames:
if n in row:
names.append(col(n))
values.append(adapt(row[n], attnames[n]))
names, values = ', '.join(names), ', '.join(values)
try:
keyname = self.pkey(table, True)
except KeyError:
raise _prg_error('Table %s has no primary key' % table)
target = ', '.join(col(k) for k in keyname)
update = []
keyname = set(keyname)
keyname.add('oid')
for n in attnames:
if n not in keyname:
value = kw.get(n, True)
if value:
if not isinstance(value, basestring):
value = 'excluded.%s' % col(n)
update.append('%s = %s' % (col(n), value))
if not values:
return row
do = 'update set %s' % ', '.join(update) if update else 'nothing'
ret = 'oid, *' if qoid else '*'
q = ('INSERT INTO %s AS included (%s) VALUES (%s)'
' ON CONFLICT (%s) DO %s RETURNING %s') % (
self._escape_qualified_name(table), names, values,
target, do, ret)
self._do_debug(q, params)
try:
q = self.db.query(q, params)
except ProgrammingError:
if self.server_version < 90500:
raise _prg_error(
'Upsert operation is not supported by PostgreSQL version')
raise # re-raise original error
res = q.dictresult()
if res: # may be empty with "do nothing"
for n, value in res[0].items():
if qoid and n == 'oid':
n = qoid
row[n] = value
else:
self.get(table, row)
return row
def clear(self, table, row=None):
"""Clear all the attributes to values determined by the types.
Numeric types are set to 0, Booleans are set to false, and everything
else is set to the empty string. If the row argument is present,
it is used as the row dictionary and any entries matching attribute
names are cleared with everything else left unchanged.
"""
# At some point we will need a way to get defaults from a table.
if row is None:
row = {} # empty if argument is not present
attnames = self.get_attnames(table)
for n, t in attnames.items():
if n == 'oid':
continue
t = t.simple
if t in DbTypes._num_types:
row[n] = 0
elif t == 'bool':
row[n] = self._make_bool(False)
else:
row[n] = ''
return row
def delete(self, table, row=None, **kw):
"""Delete an existing row in a database table.
This method deletes the row from a table. It deletes based on the
primary key of the table or the OID value as munged by get() or
passed as keyword. The OID will take precedence if provided.
The return value is the number of deleted rows (i.e. 0 if the row
did not exist and 1 if the row was deleted).
Note that if the row cannot be deleted because e.g. it is still
referenced by another table, this method raises a ProgrammingError.
"""
if table.endswith('*'): # hint for descendant tables can be ignored
table = table[:-1].rstrip()
attnames = self.get_attnames(table)
qoid = _oid_key(table) if 'oid' in attnames else None
if row is None:
row = {}
elif 'oid' in row:
del row['oid'] # only accept oid key from named args for safety
row.update(kw)
if qoid and qoid in row and 'oid' not in row:
row['oid'] = row[qoid]
if qoid and 'oid' in row: # try using the oid
keyname = ('oid',)
else: # try using the primary key
try:
keyname = self.pkey(table, True)
except KeyError: # the table has no primary key
raise _prg_error('Table %s has no primary key' % table)
# check whether all key columns have values
if not set(keyname).issubset(row):
raise KeyError('Missing value for primary key in row')
params = self.adapter.parameter_list()
adapt = params.add
col = self.escape_identifier
where = ' AND '.join('%s = %s' % (
col(k), adapt(row[k], attnames[k])) for k in keyname)
if 'oid' in row:
if qoid:
row[qoid] = row['oid']
del row['oid']
q = 'DELETE FROM %s WHERE %s' % (
self._escape_qualified_name(table), where)
self._do_debug(q, params)
res = self.db.query(q, params)
return int(res)
def truncate(self, table, restart=False, cascade=False, only=False):
"""Empty a table or set of tables.
This method quickly removes all rows from the given table or set
of tables. It has the same effect as an unqualified DELETE on each
table, but since it does not actually scan the tables it is faster.
Furthermore, it reclaims disk space immediately, rather than requiring
a subsequent VACUUM operation. This is most useful on large tables.
If restart is set to True, sequences owned by columns of the truncated
table(s) are automatically restarted. If cascade is set to True, it
also truncates all tables that have foreign-key references to any of
the named tables. If the parameter only is not set to True, all the
descendant tables (if any) will also be truncated. Optionally, a '*'
can be specified after the table name to explicitly indicate that
descendant tables are included.
"""
if isinstance(table, basestring):
only = {table: only}
table = [table]
elif isinstance(table, (list, tuple)):
if isinstance(only, (list, tuple)):
only = dict(zip(table, only))
else:
only = dict.fromkeys(table, only)
elif isinstance(table, (set, frozenset)):
only = dict.fromkeys(table, only)
else:
raise TypeError('The table must be a string, list or set')
if not (restart is None or isinstance(restart, (bool, int))):
raise TypeError('Invalid type for the restart option')
if not (cascade is None or isinstance(cascade, (bool, int))):
raise TypeError('Invalid type for the cascade option')
tables = []
for t in table:
u = only.get(t)
if not (u is None or isinstance(u, (bool, int))):
raise TypeError('Invalid type for the only option')
if t.endswith('*'):
if u:
raise ValueError(
'Contradictory table name and only options')
t = t[:-1].rstrip()
t = self._escape_qualified_name(t)
if u:
t = 'ONLY %s' % t
tables.append(t)
q = ['TRUNCATE', ', '.join(tables)]
if restart:
q.append('RESTART IDENTITY')
if cascade:
q.append('CASCADE')
q = ' '.join(q)
self._do_debug(q)
return self.db.query(q)
def get_as_list(self, table, what=None, where=None,
order=None, limit=None, offset=None, scalar=False):
"""Get a table as a list.
This gets a convenient representation of the table as a list
of named tuples in Python. You only need to pass the name of
the table (or any other SQL expression returning rows). Note that
by default this will return the full content of the table which
can be huge and overflow your memory. However, you can control
the amount of data returned using the other optional parameters.
The parameter 'what' can restrict the query to only return a
subset of the table columns. It can be a string, list or a tuple.
The parameter 'where' can restrict the query to only return a
subset of the table rows. It can be a string, list or a tuple
of SQL expressions that all need to be fulfilled. The parameter
'order' specifies the ordering of the rows. It can also be a
other string, list or a tuple. If no ordering is specified,
the result will be ordered by the primary key(s) or all columns
if no primary key exists. You can set 'order' to False if you
don't care about the ordering. The parameters 'limit' and 'offset'
can be integers specifying the maximum number of rows returned
and a number of rows skipped over.
If you set the 'scalar' option to True, then instead of the
named tuples you will get the first items of these tuples.
This is useful if the result has only one column anyway.
"""
if not table:
raise TypeError('The table name is missing')
if what:
if isinstance(what, (list, tuple)):
what = ', '.join(map(str, what))
if order is None:
order = what
else:
what = '*'
q = ['SELECT', what, 'FROM', table]
if where:
if isinstance(where, (list, tuple)):
where = ' AND '.join(map(str, where))
q.extend(['WHERE', where])
if order is None:
try:
order = self.pkey(table, True)
except (KeyError, ProgrammingError):
try:
order = list(self.get_attnames(table))
except (KeyError, ProgrammingError):
pass
if order:
if isinstance(order, (list, tuple)):
order = ', '.join(map(str, order))
q.extend(['ORDER BY', order])
if limit:
q.append('LIMIT %d' % limit)
if offset:
q.append('OFFSET %d' % offset)
q = ' '.join(q)
self._do_debug(q)
q = self.db.query(q)
res = q.namedresult()
if res and scalar:
res = [row[0] for row in res]
return res
def get_as_dict(self, table, keyname=None, what=None, where=None,
order=None, limit=None, offset=None, scalar=False):
"""Get a table as a dictionary.
This method is similar to get_as_list(), but returns the table
as a Python dict instead of a Python list, which can be even
more convenient. The primary key column(s) of the table will
be used as the keys of the dictionary, while the other column(s)
will be the corresponding values. The keys will be named tuples
if the table has a composite primary key. The rows will be also
named tuples unless the 'scalar' option has been set to True.
With the optional parameter 'keyname' you can specify an alternative
set of columns to be used as the keys of the dictionary. It must
be set as a string, list or a tuple.
If the Python version supports it, the dictionary will be an
OrderedDict using the order specified with the 'order' parameter
or the key column(s) if not specified. You can set 'order' to False
if you don't care about the ordering. In this case the returned
dictionary will be an ordinary one.
"""
if not table:
raise TypeError('The table name is missing')
if not keyname:
try:
keyname = self.pkey(table, True)
except (KeyError, ProgrammingError):
raise _prg_error('Table %s has no primary key' % table)
if isinstance(keyname, basestring):
keyname = [keyname]
elif not isinstance(keyname, (list, tuple)):
raise KeyError('The keyname must be a string, list or tuple')
if what:
if isinstance(what, (list, tuple)):
what = ', '.join(map(str, what))
if order is None:
order = what
else:
what = '*'
q = ['SELECT', what, 'FROM', table]
if where:
if isinstance(where, (list, tuple)):
where = ' AND '.join(map(str, where))
q.extend(['WHERE', where])
if order is None:
order = keyname
if order:
if isinstance(order, (list, tuple)):
order = ', '.join(map(str, order))
q.extend(['ORDER BY', order])
if limit:
q.append('LIMIT %d' % limit)
if offset:
q.append('OFFSET %d' % offset)
q = ' '.join(q)
self._do_debug(q)
q = self.db.query(q)
res = q.getresult()
cls = OrderedDict if order else dict
if not res:
return cls()
keyset = set(keyname)
fields = q.listfields()
if not keyset.issubset(fields):
raise KeyError('Missing keyname in row')
keyind, rowind = [], []
for i, f in enumerate(fields):
(keyind if f in keyset else rowind).append(i)
keytuple = len(keyind) > 1
getkey = itemgetter(*keyind)
keys = map(getkey, res)
if scalar:
rowind = rowind[:1]
rowtuple = False
else:
rowtuple = len(rowind) > 1
if scalar or rowtuple:
getrow = itemgetter(*rowind)
else:
rowind = rowind[0]
getrow = lambda row: (row[rowind],)
rowtuple = True
rows = map(getrow, res)
if keytuple or rowtuple:
namedresult = get_namedresult()
if namedresult:
if keytuple:
keys = namedresult(_MemoryQuery(keys, keyname))
if rowtuple:
fields = [f for f in fields if f not in keyset]
rows = namedresult(_MemoryQuery(rows, fields))
return cls(zip(keys, rows))
def notification_handler(self,
event, callback, arg_dict=None, timeout=None, stop_event=None):
"""Get notification handler that will run the given callback."""
return NotificationHandler(self,
event, callback, arg_dict, timeout, stop_event)
# if run as script, print some information
if __name__ == '__main__':
print('PyGreSQL version' + version)
print('')
print(__doc__)
|