本文整理汇总了Python中sqlalchemy.dialects.postgresql.insert函数的典型用法代码示例。如果您正苦于以下问题:Python insert函数的具体用法?Python insert怎么用?Python insert使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了insert函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_on_conflict_do_nothing_connectionless
def test_on_conflict_do_nothing_connectionless(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
result = conn.execute(
insert(users).on_conflict_do_nothing(
constraint="uq_login_email"
),
dict(name="name1", login_email="email1"),
)
eq_(result.inserted_primary_key, [1])
eq_(result.returned_defaults, (1,))
result = testing.db.execute(
insert(users).on_conflict_do_nothing(constraint="uq_login_email"),
dict(name="name2", login_email="email1"),
)
eq_(result.inserted_primary_key, None)
eq_(result.returned_defaults, None)
eq_(
testing.db.execute(
users.select().where(users.c.id == 1)
).fetchall(),
[(1, "name1", "email1", None)],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:26,代码来源:test_on_conflict.py
示例2: get_game_id
def get_game_id(self):
if self.game_override is None:
game = twitch.get_game_playing()
if game is None:
return None
game_id, game_name = game["_id"], game["name"]
games = self.metadata.tables["games"]
with self.engine.begin() as conn:
game_data.lock_tables(conn, self.metadata)
old_id = conn.execute(sqlalchemy.select([games.c.id]).where(games.c.name == game_name)).first()
if old_id is None:
query = insert(games)
query = query.on_conflict_do_update(index_elements=[games.c.id], set_={
"name": query.excluded.name,
})
conn.execute(query, {
"id": game_id,
"name": game_name
})
else:
old_id, = old_id
conn.execute(insert(games).on_conflict_do_nothing(index_elements=[games.c.id]), {
"id": game_id,
"name": "__LRRBOT_TEMP_GAME_%s__" % game_name,
})
game_data.merge_games(conn, self.metadata, old_id, game_id, game_id)
conn.execute(games.update().where(games.c.id == game_id), {
"name": game_name,
})
return game_id
return self.game_override
开发者ID:andreasots,项目名称:lrrbot,代码行数:32,代码来源:main.py
示例3: test_on_conflict_do_nothing_target
def test_on_conflict_do_nothing_target(self):
users = self.tables.users
with testing.db.connect() as conn:
result = conn.execute(
insert(users).on_conflict_do_nothing(
index_elements=users.primary_key.columns
),
dict(id=1, name="name1"),
)
eq_(result.inserted_primary_key, [1])
eq_(result.returned_defaults, None)
result = conn.execute(
insert(users).on_conflict_do_nothing(
index_elements=users.primary_key.columns
),
dict(id=1, name="name2"),
)
eq_(result.inserted_primary_key, [1])
eq_(result.returned_defaults, None)
eq_(
conn.execute(users.select().where(users.c.id == 1)).fetchall(),
[(1, "name1")],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:26,代码来源:test_on_conflict.py
示例4: test_on_conflict_do_update_special_types_in_set
def test_on_conflict_do_update_special_types_in_set(self):
bind_targets = self.tables.bind_targets
with testing.db.connect() as conn:
i = insert(bind_targets)
conn.execute(i, {"id": 1, "data": "initial data"})
eq_(
conn.scalar(sql.select([bind_targets.c.data])),
"initial data processed"
)
i = insert(bind_targets)
i = i.on_conflict_do_update(
index_elements=[bind_targets.c.id],
set_=dict(data="new updated data")
)
conn.execute(
i, {"id": 1, "data": "new inserted data"}
)
eq_(
conn.scalar(sql.select([bind_targets.c.data])),
"new updated data processed"
)
开发者ID:rlugojr,项目名称:sqlalchemy,代码行数:25,代码来源:test_on_conflict.py
示例5: test_on_conflict_do_nothing
def test_on_conflict_do_nothing(self):
users = self.tables.users
with testing.db.connect() as conn:
conn.execute(
insert(users).on_conflict_do_nothing(),
dict(id=1, name='name1')
)
conn.execute(
insert(users).on_conflict_do_nothing(),
dict(id=1, name='name2')
)
eq_(
conn.execute(users.select().where(users.c.id == 1)).fetchall(),
[(1, 'name1')]
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:16,代码来源:test_on_conflict.py
示例6: test_on_conflict_do_update_exotic_targets_three
def test_on_conflict_do_update_exotic_targets_three(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
self._exotic_targets_fixture(conn)
# try unique constraint: cause an upsert on target
# login_email, not id
i = insert(users)
i = i.on_conflict_do_update(
constraint=self.unique_constraint,
set_=dict(id=i.excluded.id, name=i.excluded.name,
login_email=i.excluded.login_email)
)
# note: lets_index_this value totally ignored in SET clause.
conn.execute(i, dict(
id=42, name='nameunique',
login_email='[email protected]', lets_index_this='unique')
)
eq_(
conn.execute(
users.select().
where(users.c.login_email == '[email protected]')
).fetchall(),
[(42, 'nameunique', '[email protected]', 'not')]
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:26,代码来源:test_on_conflict.py
示例7: test_on_conflict_do_update_no_row_actually_affected
def test_on_conflict_do_update_no_row_actually_affected(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
self._exotic_targets_fixture(conn)
i = insert(users)
i = i.on_conflict_do_update(
index_elements=[users.c.login_email],
set_=dict(name="new_name"),
where=(i.excluded.name == "other_name"),
)
result = conn.execute(
i, dict(name="name2", login_email="[email protected]")
)
eq_(result.returned_defaults, None)
eq_(result.inserted_primary_key, None)
eq_(
conn.execute(users.select()).fetchall(),
[
(1, "name1", "[email protected]", "not"),
(2, "name2", "[email protected]", "not"),
],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:25,代码来源:test_on_conflict.py
示例8: test_on_conflict_do_update_exotic_targets_five
def test_on_conflict_do_update_exotic_targets_five(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
self._exotic_targets_fixture(conn)
# try bogus index
i = insert(users)
i = i.on_conflict_do_update(
index_elements=self.bogus_index.columns,
index_where=self.bogus_index.dialect_options["postgresql"][
"where"
],
set_=dict(
name=i.excluded.name, login_email=i.excluded.login_email
),
)
assert_raises(
exc.ProgrammingError,
conn.execute,
i,
dict(
id=1,
name="namebogus",
login_email="[email protected]",
lets_index_this="bogus",
),
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:28,代码来源:test_on_conflict.py
示例9: test_on_conflict_do_update_multivalues
def test_on_conflict_do_update_multivalues(self):
users = self.tables.users
with testing.db.connect() as conn:
conn.execute(users.insert(), dict(id=1, name="name1"))
conn.execute(users.insert(), dict(id=2, name="name2"))
i = insert(users)
i = i.on_conflict_do_update(
index_elements=users.primary_key.columns,
set_=dict(name="updated"),
where=(i.excluded.name != "name12"),
).values(
[
dict(id=1, name="name11"),
dict(id=2, name="name12"),
dict(id=3, name="name13"),
dict(id=4, name="name14"),
]
)
result = conn.execute(i)
eq_(result.inserted_primary_key, [None])
eq_(result.returned_defaults, None)
eq_(
conn.execute(users.select().order_by(users.c.id)).fetchall(),
[(1, "updated"), (2, "name2"), (3, "name13"), (4, "name14")],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:29,代码来源:test_on_conflict.py
示例10: set_game_name
def set_game_name(lrrbot, conn, event, respond_to, name):
"""
Command: !game display NAME
Section: info
eg. !game display Resident Evil: Man Fellating Giraffe
Change the display name of the current game to NAME.
"""
game_id = lrrbot.get_game_id()
if game_id is None:
conn.privmsg(respond_to, "Not currently playing any game.")
return
show_id = lrrbot.get_show_id()
games = lrrbot.metadata.tables["games"]
game_per_show_data = lrrbot.metadata.tables["game_per_show_data"]
with lrrbot.engine.begin() as pg_conn:
name_query = sqlalchemy.select([games.c.name]).where(games.c.id == game_id)
# NULLIF: https://www.postgresql.org/docs/9.6/static/functions-conditional.html#FUNCTIONS-NULLIF
query = insert(game_per_show_data).values({
"game_id": game_id,
"show_id": show_id,
'display_name': sqlalchemy.func.nullif(name, name_query),
})
query = query.on_conflict_do_update(
index_elements=[game_per_show_data.c.game_id, game_per_show_data.c.show_id],
set_={
'display_name': query.excluded.display_name,
}
)
pg_conn.execute(query)
real_name, = pg_conn.execute(name_query).first()
conn.privmsg(respond_to, "OK, I'll start calling %s \"%s\"" % (real_name, name))
开发者ID:andreasots,项目名称:lrrbot,代码行数:35,代码来源:game.py
示例11: test_on_conflict_do_update_exotic_targets_two
def test_on_conflict_do_update_exotic_targets_two(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
self._exotic_targets_fixture(conn)
# try primary key constraint: cause an upsert on unique id column
i = insert(users)
i = i.on_conflict_do_update(
index_elements=users.primary_key.columns,
set_=dict(
name=i.excluded.name, login_email=i.excluded.login_email
),
)
result = conn.execute(
i,
dict(
id=1,
name="name2",
login_email="[email protected]",
lets_index_this="not",
),
)
eq_(result.inserted_primary_key, [1])
eq_(result.returned_defaults, None)
eq_(
conn.execute(users.select().where(users.c.id == 1)).fetchall(),
[(1, "name2", "[email protected]", "not")],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:29,代码来源:test_on_conflict.py
示例12: external_clips_save
def external_clips_save(session):
ext_channel = server.db.metadata.tables["external_channel"]
ext_vids = server.db.metadata.tables["external_video"]
if flask.request.values['action'] == "videos":
# Save video selection
with server.db.engine.begin() as conn:
videos = []
for video in flask.request.values.getlist('selected'):
chanid, vodid = video.split('-', 1)
videos.append({
"channel": int(chanid),
"vodid": vodid,
})
query = insert(ext_vids).on_conflict_do_nothing(index_elements=[ext_vids.c.vodid])
conn.execute(query, videos)
conn.execute(ext_vids.delete().where(ext_vids.c.vodid.notin_(v['vodid'] for v in videos)))
return flask.redirect(flask.url_for('clips_vidlist'), code=303)
elif flask.request.values['action'] == "add":
# Add a new channel
channel = get_user(name=flask.request.values['channel'])
with server.db.engine.begin() as conn:
conn.execute(ext_channel.insert(),
channel=channel.name)
return flask.redirect(flask.url_for('external_clips'), code=303)
elif flask.request.values['action'] == "remove":
channel = int(flask.request.values['channel'])
with server.db.engine.begin() as conn:
conn.execute(ext_channel.delete().where(ext_channel.c.id == channel))
return flask.redirect(flask.url_for('external_clips'), code=303)
else:
raise ValueError("Unexpected mode %r" % flask.request.values['action'])
开发者ID:andreasots,项目名称:lrrbot,代码行数:32,代码来源:clips.py
示例13: test_on_conflict_do_update_exotic_targets_four_no_pk
def test_on_conflict_do_update_exotic_targets_four_no_pk(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
self._exotic_targets_fixture(conn)
# try unique constraint by name: cause an
# upsert on target login_email, not id
i = insert(users)
i = i.on_conflict_do_update(
index_elements=[users.c.login_email],
set_=dict(
id=i.excluded.id,
name=i.excluded.name,
login_email=i.excluded.login_email,
),
)
result = conn.execute(
i, dict(name="name3", login_email="[email protected]")
)
eq_(result.inserted_primary_key, [1])
eq_(result.returned_defaults, (1,))
eq_(
conn.execute(users.select().order_by(users.c.id)).fetchall(),
[
(1, "name3", "[email protected]", "not"),
(2, "name2", "[email protected]", "not"),
],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:30,代码来源:test_on_conflict.py
示例14: test_on_conflict_do_update_exotic_targets_four
def test_on_conflict_do_update_exotic_targets_four(self):
users = self.tables.users_xtra
with testing.db.connect() as conn:
self._exotic_targets_fixture(conn)
# try unique constraint by name: cause an
# upsert on target login_email, not id
i = insert(users)
i = i.on_conflict_do_update(
constraint=self.unique_constraint.name,
set_=dict(
id=i.excluded.id, name=i.excluded.name,
login_email=i.excluded.login_email)
)
# note: lets_index_this value totally ignored in SET clause.
result = conn.execute(i, dict(
id=43, name='nameunique2',
login_email='[email protected]', lets_index_this='unique')
)
eq_(result.inserted_primary_key, [43])
eq_(result.returned_defaults, None)
eq_(
conn.execute(
users.select().
where(users.c.login_email == '[email protected]')
).fetchall(),
[(43, 'nameunique2', '[email protected]', 'not')]
)
开发者ID:rlugojr,项目名称:sqlalchemy,代码行数:30,代码来源:test_on_conflict.py
示例15: _exotic_targets_fixture
def _exotic_targets_fixture(self, conn):
users = self.tables.users_xtra
conn.execute(
insert(users),
dict(
id=1,
name="name1",
login_email="[email protected]",
lets_index_this="not",
),
)
conn.execute(
users.insert(),
dict(
id=2,
name="name2",
login_email="[email protected]",
lets_index_this="not",
),
)
eq_(
conn.execute(users.select().where(users.c.id == 1)).fetchall(),
[(1, "name1", "[email protected]", "not")],
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:26,代码来源:test_on_conflict.py
示例16: insert_horse
def insert_horse(self, item):
stmt = insert(Horse).values(
**item
).on_conflict_do_update(
constraint='horses_pkey',
set_=dict(item)
)
self.conn.execute(stmt)
开发者ID:otomarukanta,项目名称:blacktide,代码行数:8,代码来源:pipelines.py
示例17: insert_jockey
def insert_jockey(self, item):
stmt = insert(Jockey).values(
**item
).on_conflict_do_update(
constraint='jockeys_pkey',
set_=dict(item)
)
self.conn.execute(stmt)
开发者ID:otomarukanta,项目名称:blacktide,代码行数:8,代码来源:pipelines.py
示例18: add_label
def add_label(label: str, session_id: int):
session = Session.query.get_or_404(session_id)
db.session.execute(insert(Label).values(name=label).on_conflict_do_nothing())
db.session.commit()
session.labels.append(Label.query.filter_by(name=label).one())
db.session.add(session)
db.session.commit()
开发者ID:getslash,项目名称:backslash,代码行数:9,代码来源:labels.py
示例19: test_bad_args
def test_bad_args(self):
assert_raises(
ValueError,
insert(self.tables.users).on_conflict_do_nothing,
constraint='id', index_elements=['id']
)
assert_raises(
ValueError,
insert(self.tables.users).on_conflict_do_update,
constraint='id', index_elements=['id']
)
assert_raises(
ValueError,
insert(self.tables.users).on_conflict_do_update, constraint='id'
)
assert_raises(
ValueError,
insert(self.tables.users).on_conflict_do_update
)
开发者ID:rlugojr,项目名称:sqlalchemy,代码行数:19,代码来源:test_on_conflict.py
示例20: patreon_login
async def patreon_login(session):
code = flask.request.args.get('code')
state_param = flask.request.args.get('state')
state_sess = flask.session.pop('patreon_state', None)
if code is None or state_param is None or state_sess is None:
flask.flash('OAuth parameters missing', 'error')
return flask.redirect(flask.url_for('patreon_index'))
if state_param != state_sess:
flask.flash('Nonce mismatch: %r not equal to %r' % (state_param, state_sess), 'error')
return flask.redirect(flask.url_for('patreon_index'))
access_token, refresh_token, expiry = await patreon.request_token('authorization_code',
code=code,
redirect_uri=config['patreon_redirect_uri'],
)
user = await patreon.current_user(access_token)
patreon_users = server.db.metadata.tables['patreon_users']
users = server.db.metadata.tables['users']
pledge_start = None
for pledge in user['data'].get('relationships', {}).get('pledges', {}).get('data', []):
for obj in user['included']:
if obj['type'] == pledge['type'] and obj['id'] == pledge['id'] and obj['attributes']['amount_cents'] > 0:
pledge_start = dateutil.parser.parse(obj['attributes']['created_at'])
break
else:
continue
break
with server.db.engine.begin() as conn:
query = insert(patreon_users).returning(patreon_users.c.id)
query = query.on_conflict_do_update(
index_elements=[patreon_users.c.patreon_id],
set_={
'full_name': query.excluded.full_name,
'access_token': query.excluded.access_token,
'refresh_token': query.excluded.refresh_token,
'token_expires': query.excluded.token_expires,
'pledge_start': query.excluded.pledge_start,
}
)
patreon_user_id, = conn.execute(query,
patreon_id=user['data']['id'],
full_name=user['data']['attributes']['full_name'],
access_token=access_token,
refresh_token=refresh_token,
token_expires=expiry,
pledge_start=pledge_start,
).first()
conn.execute(users.update().where(users.c.id == session['user']['id']), patreon_user_id=patreon_user_id)
flask.flash('Patreon account linked.', 'success')
return flask.redirect(flask.url_for('patreon_index'))
开发者ID:andreasots,项目名称:lrrbot,代码行数:55,代码来源:patreon.py
注:本文中的sqlalchemy.dialects.postgresql.insert函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论