Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- 1)
- from sqlalchemy import create_engine
- engine = create_engine("sqlite:///some.db")
- #создаем таблицу
- engine.execute("""CREATE TABLE employee (
- emp_id integer primary key,
- emp_name varchar(30)
- )""")
- #Вставляем строку с emp_name = 'dilbert'
- engine.execute("""INSERT into employee(emp_name) values ('dilbert')""")
- #выбираем все строки и столбцы таблицы
- result = engine.execute("SELECT * from employee")
- #выводим
- print(result.fetchall())
- Результат: [(1, 'dilbert')]
- _________________________________________________________
- 2)
- __13/20__
- from sqlalchemy import create_engine
- from sqlalchemy import MetaData
- from sqlalchemy import Table, Column
- from sqlalchemy import Integer, String, DateTime, ForeignKey
- engine = create_engine("sqlite:///some.db")
- metadata = MetaData()
- #таблицы
- 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', ForeignKey('user.id'))
- )
- user_table = Table('user', metadata,
- Column('id', Integer, primary_key=True),
- Column('name', String(30), nullable=False),
- Column('fullname', String(100), nullable=False)
- )
- metadata.create_all(engine)
- __20/20__
- from sqlalchemy import create_engine
- from sqlalchemy import MetaData
- from sqlalchemy import Table, Column
- from sqlalchemy import Integer, String, DateTime, ForeignKey
- engine = create_engine("sqlite:///some.db")
- metadata = MetaData()
- #таблицы
- 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', ForeignKey('user.id'))
- )
- user_table = Table('user', metadata,
- Column('id', Integer, primary_key=True),
- Column('name', String(30), nullable=False),
- Column('fullname', String(100), nullable=False)
- )
- metadata.create_all(engine)
- from sqlalchemy import inspect
- metadata2 = MetaData()
- network_reflected = Table('network', metadata2, autoload=True, autoload_with=engine)
- print(network_reflected.c)
- inspector = inspect(engine)
- #таблицы с колонкой story_id
- print("Tables with story_id:")
- for mytables in inspector.get_table_names():
- for column in inspector.get_columns(mytables):
- if column ['name'] == 'story_id':
- print (mytables)
- Результат:
- ['network.network_id', 'network.name', 'network.created_at', 'network.owner_id']
- Tables with story_id
- _________________________________________________________
- 3)
- from sqlalchemy import create_engine
- from sqlalchemy import MetaData
- from sqlalchemy import Table, Column
- from sqlalchemy import Integer, String, ForeignKey
- from sqlalchemy import select
- from sqlalchemy import func
- metaData = MetaData()
- engine = create_engine("sqlite:///some.db")
- user_table = Table('user', metaData,
- Column('id', Integer, primary_key=True),
- Column('username', String(50)),
- Column('fullname', String(50))
- )
- metaData.create_all(engine)
- # __18/46__
- print(user_table.c.fullname == 'ed')
- print((user_table.c.fullname == 'ed') & (user_table.c.id > 5))
- print((user_table.c.username == 'edward') | ((user_table.c.fullname == 'ed') & (user_table.c.id > 5)))
- Результат:
- "user".fullname = :fullname_1
- "user".fullname = :fullname_1 AND "user".id > :id_1
- "user".username = :username_1 OR "user".fullname = :fullname_1 AND "user".id > :id_1
- #__27/46__
- data = user_table.insert().values(username='dilbert', fullname='Dilbert Jones')
- connection = engine.connect()
- result = connection.execute(data)
- print(result)
- connection.close()
- select_data=select([user_table.c.id, user_table.c.username, user_table.c.fullname]).\
- where((user_table.c.username == 'wendy')|(user_table.c.username == 'dilbert')).\
- order_by(user_table.c.fullname)
- result=engine.execute(select_data)
- print(result.fetchall())
- Результат:
- [(1, 'dilbert', 'Dilbert Jones')]
- #__38/46__
- address_table = Table("address", metaData,
- Column('id', Integer, primary_key=True),
- Column('user_id', Integer, ForeignKey('user.id'), nullable=False),
- Column('email_address', String(100), nullable=False)
- )
- metaData.create_all(engine)
- engine.execute(user_table.insert().values(username='ed', fullname='Ed Jones'))
- engine.execute(address_table.insert().values(user_id='1', email_address='ed@ed.com'))
- engine.execute(address_table.insert().values(user_id='1', email_address='ed@gmail.com'))
- engine.execute(address_table.insert().values(user_id='2', email_address='jack@yahoo.com'))
- engine.execute(address_table.insert().values(user_id='3', email_address='wendy@gmail.com'))
- selected_data=select([user_table.c.fullname, address_table.c.email_address]).select_from(user_table.join(address_table)).\
- where(user_table.c.username == 'ed').order_by(address_table.c.email_address)
- result=engine.execute(selected_data)
- print(result.fetchall())
- Результат: [('Ed Jones', 'ed@ed.com'), ('Ed Jones', 'ed@gmail.com')]
- #__46/46__
- result=engine.execute(user_table.update().values(fullname="Ed Jones").where(user_table.c.username == "ed"))
- select_address = select([address_table.c.email_address]).\
- where(user_table.c.id == address_table.c.user_id)
- result =select([user_table.c.username, select_address.as_scalar().where(user_table.c.username.in_(['jack', 'wendy']))])
- print(engine.execute(result).fetchall())
- Обновили 1 колонку.
- Результат: [('ed', None)]
- _________________________________________________________
- 4)
- from sqlalchemy import create_engine, func, select
- from sqlalchemy import MetaData,Column, Integer, String, Numeric, ForeignKey
- from sqlalchemy.orm import Session, relationship
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.orm import aliased
- metaData = MetaData()
- engine = create_engine("sqlite:///some.db")
- base = declarative_base()
- class Network(base):
- __tablename__ = 'network'
- 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(engine)
- #__25/72__
- session = Session(bind=engine)
- session.add_all([Network(name='net1'), Network(name='net2')])
- session.commit()
- #__43/72__
- session.add_all([
- User(name='ed', fullname='Ed Jones'),
- User(name='wendy', fullname='Wendy Weathersmith'),
- User(name='mary', fullname='Mary Contrary'),
- User(name='fred', fullname='Fred Flinstone')
- ])
- session.commit()
- query = session.query(User.fullname).order_by(User.fullname)
- print(query.all())
- query = query.filter(User.name.in_(['mary','ed']))
- print(query.all())
- Результат:
- [('Ed Jones',), ('Fred Flinstone',), ('Mary Contrary',), ('Wendy Weathersmith',)]
- [('Ed Jones',), ('Mary Contrary',)]
- ('Mary Contrary',)
- print(query[1])
- #__62/72__
- jack = User(name='jack', fullname='Jack Bean')
- jack.addresses = [Address(email_address='jack@gmail.com'),
- Address(email_address='j25@yahoo.com'),
- Address(email_address='jack@hotmail.com')]
- session.add(jack)
- session.commit()
- query = session.query(User.name, Address.email_address).join(Address).filter(Address.email_address=="j25@yahoo.com")
- print (query.all())
- user1 = aliased(User)
- user2 = aliased(User)
- query=engine.execute(select([user1.name, user2.name]).distinct().where(user1.name < user2.name))
- print(query.fetchall())
- Результат:
- [('jack', 'j25@yahoo.com')]
- [('ed', 'wendy'), ('ed', 'mary'), ('ed', 'fred'), ('ed', 'jack'), ('mary', 'wendy'), ('fred', 'wendy'), ('fred', 'mary'), ('fred', 'jack'), ('jack', 'wendy'), ('jack', 'mary')]
- #__72/72__
- a1=Account(owner = "Jack Jones", balance = 5000)
- a2=Account(owner="Ed Rendell", balance=10000)
- session.add_all([
- Transaction(amount=500, account=a1),
- Transaction(amount=4500, account=a1),
- Transaction(amount=6000, account=a2),
- Transaction(amount=4000, account=a2),
- a1,
- a2,
- ])
- session.commit()
- for account in session.query(Account).all():
- print(account.owner)
- print(account.balance)
- print(func.sum(Transaction.amount).label('summed')
- ).select_from(Account).join(Transaction.account).all())
Add Comment
Please, Sign In to add comment