from sqlalchemy import create_engine
import os
if os.path.exists("database.db"):
os.remove("database.db")
table = create_engine("sqlite:///database.db")
table.execute("""
create table employee (
emp_id integer primary key,
emp_name varchar
)
""")
table.execute("insert into employee (emp_name) values (:emp_name)", emp_name="dilbert")
print(table.execute("select * from employee").fetchall())
[(1, 'dilbert')]
from sqlalchemy import *
database = create_engine("sqlite:///database.db")
metaData=MetaData()
user_table = Table('user', metaData,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('fullname', String)
)
addresses_table = Table('address', metaData,
Column('id', Integer, primary_key=True),
Column('email_address', String(100), nullable=False),
Column('user_id', Integer, ForeignKey('user.id'))
)
story_table = Table('story', metaData,
Column('story_id', Integer, primary_key=True),
Column('version_id', Integer, primary_key=True),
Column('headline', Unicode(100), nullable=False),
Column('body', UnicodeText)
)
published_table = Table('published', metaData,
Column('pub_id', Integer, primary_key=True),
Column('pub_timestamp', DateTime, nullable=False),
Column('story_id', Integer),
Column('version_id', Integer),
ForeignKeyConstraint(
['story_id', 'version_id'],
['story.story_id', 'story.version_id'])
)
network_table= Table('network',metaData,
Column('network_id', Integer, primary_key=True),
Column('name', String(100), nullable=False),
Column('created_at', DateTime, nullable=False),
Column('owner_id', Integer,ForeignKey('user.id'))
)
metaData.create_all(database)
metadata2=MetaData()
reflect = Table('user', metadata2, autoload=True, autoload_with=database)
inspector = inspect(database)
columns=inspector.get_columns('network')
for column in columns:
print(column)
tableNames=inspector.get_table_names()
result = []
for tableName in tableNames:
for column in inspector.get_columns(tableName):
if column['name'] == 'story_id':
result.append(tableName)
print(result)
{'default': None, 'nullable': False, 'name': 'network_id', 'autoincrement': True, 'type': INTEGER(), 'primary_key': 1}
{'default': None, 'nullable': False, 'name': 'name', 'autoincrement': True, 'type': VARCHAR(length=100), 'primary_key': 0}
{'default': None, 'nullable': False, 'name': 'created_at', 'autoincrement': True, 'type': DATETIME(), 'primary_key': 0}
{'default': None, 'nullable': True, 'name': 'owner_id', 'autoincrement': True, 'type': INTEGER(), 'primary_key': 0}
['published', 'story']
from sqlalchemy import *
metadata = MetaData()
database = create_engine("sqlite:///database.db")
connection = database.connect()
userTable = Table('username', metadata,
Column('id', Integer, primary_key=True),
Column('username', String(50)),
Column('fullname', String(50))
)
print(userTable.columns)
addressTable = Table("address", metadata,
Column('id', Integer, primary_key=True),
Column('username_id', Integer, ForeignKey('username.id'),
nullable=False),
Column('email_address', String(100), nullable=False)
)
metadata.create_all(database)
connection.execute(addressTable.insert(), [
{"user_id": 1, "email_address": "ed@ed.com"},
{"user_id": 1, "email_address": "ed@gmail.com"},
{"user_id": 2, "email_address": "jack@yahoo.com"},
{"user_id": 3, "email_address": "wendy@gmail.com"},
])
print(userTable.c.fullname == 'ed')
print(and_(userTable.c.fullname == 'ed',userTable.c.id > 5))
print(or_(userTable.c.username == 'ed'))
result = connection.execute(userTable.insert().values(username='ed', fullname='Ed Jones'))
result.inserted_primary_key
connection.execute(userTable.insert(), [
{'username': 'jack', 'fullname': 'Jack Burger'},
{'username': 'wendy', 'fullname': 'Wendy Weathersmith'}
])
result = connection.execute(userTable.insert(), [
{'username': 'dilbert',
'fullname': 'Dilbert Jones'}
])
print(result.inserted_primary_key)
selecti = select([userTable]).\
where(or_(userTable.c.username == 'wendy',userTable.c.username == 'dilbert')).\
order_by(userTable.c.fullname)
print(connection.execute(selecti).fetchall())
text = select([userTable.c.fullname, addressTable.c.email_address]).\
select_from(userTable.join(addressTable)).\
where(userTable.c.username=='ed').\
order_by(addressTable.c.email_address)
print(text)
result = userTable.update().values(fullname="Ed Jones").where(userTable.c.username == 'ed')
print(result)
connection.execute(userTable.delete())
connection.execute(addressTable.delete())
['username.id', 'username.username', 'username.fullname']
username.fullname = :fullname_1
username.fullname = :fullname_1 AND username.id > :id_1
username.username = :username_1
[20]
[(4, 'dilbert', 'Dilbert Jones'), (8, 'dilbert', 'Dilbert Jones'), (12, 'dilbert', 'Dilbert Jones'), (16, 'dilbert', 'Dilbert Jones'), (20, 'dilbert', 'Dilbert Jones'), (3, 'wendy', 'Wendy Weathersmith'), (7, 'wendy', 'Wendy Weathersmith'), (11, 'wendy', 'Wendy Weathersmith'), (15, 'wendy', 'Wendy Weathersmith'), (19, 'wendy', 'Wendy Weathersmith')]
SELECT username.fullname, address.email_address
FROM username JOIN address ON username.id = address.username_id
WHERE username.username = :username_1 ORDER BY address.email_address
UPDATE username SET fullname=:fullname WHERE username.username = :username_1
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session,relationship
Base = declarative_base()
database = create_engine("sqlite:///some.db")
connection=database.connect()
class Network(Base):
__tablename__ = 'network'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
def __repr__(self):
return "Network: %r" % (self.name)
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
name = Column(String)
fullname = Column(String)
def __repr__(self):
return "<User(%r, %r)>" % (self.name, self.fullname)
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
email_address = Column(String, nullable=False)
user_id = Column(Integer, ForeignKey('user.id'))
user = relationship("User", backref="addresses")
def __repr__(self):
return "<Address(%r)>" % self.email_address
class Account(Base):
__tablename__ = 'account'
id = Column(Integer, primary_key=True)
owner = Column(String(50), nullable=False)
balance = Column(Numeric, default=0)
def __repr__(self):
return "Account: %r, %r" % (self.owner, self.balance)
class Transaction(Base):
__tablename__ = 'transaction'
id = Column(Integer, primary_key=True)
amount = Column(Numeric, nullable=False)
account_id = Column(Integer, ForeignKey(Account.__tablename__ + '.id'), nullable=False)
account = relationship('Account', backref="transactions")
def __repr__(self):
return "Transaction: %r" % (self.amount)
Base.metadata.create_all(database)
ses1 = Session(bind = database)
ses1.add(Network(name = 'net1'))
ses1.add(Network(name = 'net2'))
ses1.commit()
print(database.execute("select * from network").fetchall())
ses2 = Session(bind = database)
ses2.add(User(name='wendy', fullname='Wendy Weathersmith'))
ses2.add(User(name='mary', fullname='Mary Contrary'))
ses2.add(User(name='fred', fullname='Fred Flinstone'))
query = ses2.query(User.fullname).order_by(User.fullname)
print(query)
query2 = query.filter(User.name.in_(['mary', 'ed']))
print(query2)
query3 = ses2.query(User.name, Address.email_address).join(Address).filter(Address.email_address.in_(['j25@yahoo.com']))
print(query3)
ses3=Session(bind = database)
acc1=Account(owner = "Jack Jones", balance = 5000)
acc2=Account(owner="Ed Rendell", balance=10000)
ses3.add_all([Account(owner = "Jack Jones", balance = 5000),
Account(owner="Ed Rendell", balance=10000),
Transaction(amount = 500, account=acc1),
Transaction(amount = 4500, account=acc1),
Transaction(amount = 6000, account=acc2),
Transaction(amount = 4000, account = acc2)])
for account in ses3.query(Account).all():
owner = account.owner
balance = account.balance
spent_money = 0
for account_transaction in account.transactions:
spent_money += account_transaction.amount
print("Account owner: " + str(owner) + '\t' +
"Account balance: " + str(balance) + '\t' +
"Spent money: " + str(spent_money))
[(1, 'net1'), (2, 'net2'), (3, 'net1'), (4, 'net2'), (5, 'net1'), (6, 'net2'), (7, 'net1'), (8, 'net2')]
SELECT "user".fullname AS user_fullname
FROM "user" ORDER BY "user".fullname
SELECT "user".fullname AS user_fullname
FROM "user"
WHERE "user".name IN (:name_1, :name_2) ORDER BY "user".fullname
SELECT "user".name AS user_name, address.email_address AS address_email_address
FROM "user" JOIN address ON "user".id = address.user_id
WHERE address.email_address IN (:email_address_1)
Account owner: Jack Jones Account balance: 5000.0000000000 Spent money: 0
Account owner: Ed Rendell Account balance: 10000.0000000000 Spent money: 0
Account owner: Jack Jones Account balance: 5000 Spent money: 5000
Account owner: Ed Rendell Account balance: 10000 Spent money: 10000