Python源码示例：asyncpg.Record()
示例1
def _get_comment_from_db_record(
self,
*,
comment_row: Record,
author_username: str,
requested_user: Optional[User],
) -> Comment:
return Comment(
id_=comment_row["id"],
body=comment_row["body"],
author=await self._profiles_repo.get_profile_by_username(
username=author_username, requested_user=requested_user,
),
created_at=comment_row["created_at"],
updated_at=comment_row["updated_at"],
)
示例2
def test_record_gc(self):
elem = object()
mapping = {}
with self.checkref(mapping, elem):
r = Record(mapping, (elem,))
del r
key = 'spam'
val = int('101010')
mapping = {key: val}
with self.checkref(key, val):
r = Record(mapping, (0,))
with self.assertRaises(RuntimeError):
r[key]
del r
key = 'spam'
val = 'ham'
mapping = {key: val}
with self.checkref(key, val):
r = Record(mapping, (0,))
with self.assertRaises(RuntimeError):
r[key]
del r
示例3
def test_record_len_getindex(self):
r = Record(R_A, (42,))
self.assertEqual(len(r), 1)
self.assertEqual(r[0], 42)
self.assertEqual(r['a'], 42)
r = Record(R_AB, (42, 43))
self.assertEqual(len(r), 2)
self.assertEqual(r[0], 42)
self.assertEqual(r[1], 43)
self.assertEqual(r['a'], 42)
self.assertEqual(r['b'], 43)
with self.assertRaisesRegex(IndexError,
'record index out of range'):
r[1000]
with self.assertRaisesRegex(KeyError, 'spam'):
r['spam']
with self.assertRaisesRegex(KeyError, 'spam'):
Record(None, (1,))['spam']
with self.assertRaisesRegex(RuntimeError, 'invalid record descriptor'):
Record({'spam': 123}, (1,))['spam']
示例4
def test_record_hash(self):
AB = collections.namedtuple('AB', ('a', 'b'))
r1 = Record(R_AB, (42, 43))
r2 = Record(R_AB, (42, 43))
r3 = Record(R_AB, (42, 45))
r4 = (42, 43)
r5 = AB(42, 43)
self.assertEqual(hash(r1), hash(r2))
self.assertNotEqual(hash(r1), hash(r3))
self.assertEqual(hash(r1), hash(r4))
self.assertEqual(hash(r1), hash(r5))
d = {}
d[r1] = 123
self.assertEqual(d[r1], 123)
self.assertIn(r2, d)
self.assertEqual(d[r2], 123)
self.assertNotIn(r3, d)
self.assertIn(r4, d)
示例5
def _get_article_from_db_record(
self,
*,
article_row: Record,
slug: str,
author_username: str,
requested_user: Optional[User],
) -> Article:
return Article(
id_=article_row["id"],
slug=slug,
title=article_row["title"],
description=article_row["description"],
body=article_row["body"],
author=await self._profiles_repo.get_profile_by_username(
username=author_username, requested_user=requested_user,
),
tags=await self.get_tags_for_article_by_slug(slug=slug),
favorites_count=await self.get_favorites_count_for_article_by_slug(
slug=slug,
),
favorited=await self.is_article_favorited_by_user(
slug=slug, user=requested_user,
)
if requested_user
else False,
created_at=article_row["created_at"],
updated_at=article_row["updated_at"],
)
示例6
def upgrade(self, pool: asyncpg.pool.Pool) -> None:
async with pool.acquire() as conn:
await conn.execute("""CREATE TABLE IF NOT EXISTS version (
version INTEGER PRIMARY KEY
)""")
row: asyncpg.Record = await conn.fetchrow("SELECT version FROM version LIMIT 1")
version = row["version"] if row else 0
if len(self.upgrades) < version:
error = (f"Unsupported database version v{version} "
f"(latest known is v{len(self.upgrades) - 1})")
if not self.allow_unsupported:
raise UnsupportedDatabaseVersion(error)
else:
self.log.warning(error)
return
elif len(self.upgrades) == version:
self.log.debug(f"Database at v{version}, not upgrading")
return
for new_version in range(version + 1, len(self.upgrades)):
upgrade = self.upgrades[new_version]
desc = getattr(upgrade, "__mau_db_upgrade_description__", None)
suffix = f": {desc}" if desc else ""
self.log.debug(f"Upgrading database from v{version} to v{new_version}{suffix}")
await upgrade(conn)
version = new_version
async with conn.transaction():
self.log.debug(f"Saving current version (v{version}) to database")
await conn.execute("DELETE FROM version")
await conn.execute("INSERT INTO version (version) VALUES ($1)", version)
示例7
def fetch(self, query: str, *args: Any, timeout: Optional[float] = None
) -> List[asyncpg.Record]:
async with self.acquire() as conn:
return await conn.fetch(query, *args, timeout=timeout)
示例8
def fetchrow(self, query: str, *args: Any, timeout: Optional[float] = None
) -> asyncpg.Record:
async with self.acquire() as conn:
return await conn.fetchrow(query, *args, timeout=timeout)
示例9
def __init__(self, table_name, val: Record):
self.table = table_name
self.val = val # 只是为了补全
示例10
def count(self) -> asyncpg.Record:
"""Return (not animated count, animated count, total)"""
return await self.bot.pool.fetchrow(self.queries.count())
示例11
def parse_record(record: Record) -> Optional[tuple]:
"""
Parse a asyncpg Record object to a tuple of values
:param record: theasyncpg Record object
:return: the tuple of values if it's not None, else None
"""
try:
return tuple(record.values())
except AttributeError:
return None
示例12
def test_record_freelist_ok(self):
for _ in range(10000):
Record(R_A, (42,))
Record(R_AB, (42, 42,))
示例13
def test_record_slice(self):
r = Record(R_ABC, (1, 2, 3))
self.assertEqual(r[:], (1, 2, 3))
self.assertEqual(r[:1], (1,))
self.assertEqual(r[::-1], (3, 2, 1))
self.assertEqual(r[::-2], (3, 1))
self.assertEqual(r[1:2], (2,))
self.assertEqual(r[2:2], ())
示例14
def test_record_immutable(self):
r = Record(R_A, (42,))
with self.assertRaisesRegex(TypeError, 'does not support item'):
r[0] = 1
示例15
def test_record_iter(self):
r = Record(R_AB, (42, 43))
with self.checkref(r):
self.assertEqual(iter(r).__length_hint__(), 2)
self.assertEqual(tuple(r), (42, 43))
示例16
def test_record_values(self):
r = Record(R_AB, (42, 43))
vv = r.values()
self.assertEqual(tuple(vv), (42, 43))
self.assertTrue(repr(vv).startswith('<RecordIterator '))
示例17
def test_record_keys(self):
r = Record(R_AB, (42, 43))
vv = r.keys()
self.assertEqual(tuple(vv), ('a', 'b'))
self.assertEqual(list(Record(None, (42, 43)).keys()), [])
示例18
def test_record_items(self):
r = Record(R_AB, (42, 43))
self.assertEqual(dict(r), {'a': 42, 'b': 43})
self.assertEqual(
list(collections.OrderedDict(r).items()),
[('a', 42), ('b', 43)])
with self.checkref(r):
rk = r.items()
self.assertEqual(rk.__length_hint__(), 2)
self.assertEqual(next(rk), ('a', 42))
self.assertEqual(rk.__length_hint__(), 1)
self.assertEqual(next(rk), ('b', 43))
self.assertEqual(rk.__length_hint__(), 0)
with self.assertRaises(StopIteration):
next(rk)
with self.assertRaises(StopIteration):
next(rk)
self.assertEqual(rk.__length_hint__(), 0)
self.assertEqual(list(r.items()), [('a', 42), ('b', 43)])
# Check invalid records just in case
r = Record(R_A, (42, 43))
self.assertEqual(list(r.items()), [('a', 42)])
r = Record(R_AB, (42,))
self.assertEqual(list(r.items()), [('a', 42)])
# Try to iterate over exhausted items() iterator
r = Record(R_A, (42, 43))
it = r.items()
list(it)
list(it)
示例19
def test_record_get(self):
r = Record(R_AB, (42, 43))
with self.checkref(r):
self.assertEqual(r.get('a'), 42)
self.assertEqual(r.get('nonexistent'), None)
self.assertEqual(r.get('nonexistent', 'default'), 'default')
示例20
def test_record_not_pickleable(self):
r = Record(R_A, (42,))
with self.assertRaises(Exception):
pickle.dumps(r)
示例21
def test_record_empty(self):
r = Record(None, ())
self.assertEqual(r, ())
self.assertLess(r, (1,))
self.assertEqual(len(r), 0)
self.assertFalse(r)
self.assertNotIn('a', r)
self.assertEqual(repr(r), '<Record>')
self.assertEqual(str(r), '<Record>')
with self.assertRaisesRegex(KeyError, 'aaa'):
r['aaa']
self.assertEqual(dict(r.items()), {})
self.assertEqual(list(r.keys()), [])
self.assertEqual(list(r.values()), [])
示例22
def test_record_duplicate_colnames(self):
"""Test that Record handles duplicate column names."""
records_descs = [
[('a', 1)],
[('a', 1), ('a', 2)],
[('a', 1), ('b', 2), ('a', 3)],
[('a', 1), ('b', 2), ('a', 3), ('c', 4), ('b', 5)],
]
for desc in records_descs:
items = collections.OrderedDict(desc)
query = 'SELECT ' + ', '.join(
['{} as {}'.format(p[1], p[0]) for p in desc])
with self.subTest(query=query):
r = await self.con.fetchrow(query)
for idx, (field, val) in enumerate(desc):
self.assertEqual(r[idx], val)
self.assertEqual(r[field], items[field])
expected_repr = '<Record {}>'.format(
' '.join('{}={}'.format(p[0], p[1]) for p in desc))
self.assertEqual(repr(r), expected_repr)
self.assertEqual(list(r.items()), desc)
self.assertEqual(list(r.values()), [p[1] for p in desc])
self.assertEqual(list(r.keys()), [p[0] for p in desc])
示例23
def test_record_no_new(self):
"""Instances of Record cannot be directly created."""
with self.assertRaisesRegex(
TypeError, "cannot create 'asyncpg.Record' instances"):
asyncpg.Record()
示例24
def test_record_cmp(self):
AB = collections.namedtuple('AB', ('a', 'b'))
r1 = Record(R_AB, (42, 43))
r2 = Record(R_AB, (42, 43))
r3 = Record(R_AB.copy(), (42, 43))
r4 = Record(R_AB.copy(), (42, 45))
r5 = Record(R_ABC, (42, 46, 57))
r6 = Record(R_AC, (42, 43))
r7 = (42, 43)
r8 = [42, 43]
r9 = AB(42, 43)
r10 = AB(42, 44)
self.assertEqual(r1, r2)
self.assertEqual(r1, r3)
self.assertEqual(r1, r6)
self.assertEqual(r1, r7)
self.assertEqual(r1, r9)
self.assertNotEqual(r1, r4)
self.assertNotEqual(r1, r10)
self.assertNotEqual(r1, (42,))
self.assertNotEqual(r1, r5)
self.assertNotEqual(r4, r5)
self.assertNotEqual(r4, r6)
self.assertNotEqual(r6, r5)
self.assertNotEqual(r1, r8)
self.assertNotEqual(r8, r6)
self.assertLess(r1, r4)
self.assertGreater(r4, r1)
self.assertLess(r1, r5)
self.assertLess(r7, r5)
self.assertLess(r1, r10)
self.assertGreater(r5, r6)
self.assertGreater(r5, r7)
self.assertGreater(r5, r4)
with self.assertRaisesRegex(
TypeError, "unorderable|'<' not supported"):
r1 < r8
self.assertEqual(
sorted([r1, r2, r3, r4, r5, r6, r7]),
[r1, r2, r3, r6, r7, r4, r5])
