Huge News!Announcing our $40M Series B led by Abstract Ventures.Learn More
Socket
Sign inDemoInstall
Socket

aiomysql-core

Package Overview
Dependencies
Maintainers
1
Alerts
File Explorer

Advanced tools

Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

aiomysql-core

Simple framework for aiomysql

  • 0.0.9
  • PyPI
  • Socket score

Maintainers
1

Aiomysql-Core

Simple framework for aiomysql

Introduce

simple package, easy to use

aiomysql

Document

click me

Installation

pip install aiomysql-core

Simple uses

import asyncio
import aiomysql
from aiomysql_core import AioMysqlCore


async def test_example(loop):
    pool = await aiomysql.create_pool(host='', port=3306,
                                      user='', password='',
                                      db='', loop=loop)
    core = AioMysqlCore(pool=pool)
    rows = await core.query('select * from users where uid=%s', 113)
    print(rows)
    rows = await core.gener('select * from users limit 100')
    async for row in rows:
        print(row)
    row = await core.get('select * from users where uid=%(uid)s', {'uid': 113})
    print(row)
    rowcount = await core.execute_rowcount('select * from users where uid=%(uid)s', {'uid': 113})
    print(rowcount)
    pool.close()
    await pool.wait_closed()


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example(loop))

Simple SQLAlchemy uses

import asyncio
from aiomysql.sa import create_engine
from aiomysql_core import AioMysqlAlchemyCore
from sqlalchemy import Column, Integer, String, MetaData, Table

metadata = MetaData()

Test = Table('test', metadata,
             Column('id', Integer, primary_key=True),
             Column('content', String(255), server_default="")
             )


async def test_example(loop):
    config = {'user': '', 'password': '', 'db': '',
              'host': '', 'port': 3306, 'autocommit': True, 'charset': 'utf8mb4'}
    engine = await create_engine(loop=loop, **config)
    core = AioMysqlAlchemyCore(engine=engine)
    # insert
    doc = {'content': 'insert'}
    clause = Test.insert().values(**doc)
    rowcount = await core.execute_rowcount(clause)
    print(rowcount)
    # search
    clause = Test.select().where(Test.c.id == 1).limit(1)
    row = await core.get(clause)
    print(row.id, row.content)
    clause = Test.select().where(Test.c.id > 1)
    rows = await core.query(clause)
    async for row in rows:
        print(row.id, row.content)
    # update
    doc = {'content': 'update'}
    clause = Test.update().values(**doc).where(Test.c.id == 1)
    rowcount = await core.execute_rowcount(clause)
    print(rowcount)
    # delete
    clause = Test.delete().where(Test.c.id == 1)
    rowcount = await core.execute_rowcount(clause)
    print(rowcount)
    await engine.wait_closed()


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example(loop))

FAQs


Did you know?

Socket

Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.

Install

Related posts

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

Packages

npm

Stay in touch

Get open source security insights delivered straight into your inbox.


  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc