| Name | mysql-cli JSON |
| Version |
1.0.0
JSON |
| download |
| home_page | |
| Summary | A MySQL Python Client |
| upload_time | 2022-12-11 15:48:59 |
| maintainer | |
| docs_url | None |
| author | |
| requires_python | >=3.7 |
| license | |
| keywords |
mysql
python
client
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# MySQL Python Client
A MySQL python client based on [mysql-connector-python](https://github.com/mysql/mysql-connector-python), with friendly interface and full-featured client configuration.
Compare with the way which only takes table names and SQL parameters as input and assembles SQL statement internally, I personally prefer developer taking complete control of SQL, making it flexible which means you can use whatever SQL feature as you need, clear which means the final statement is obviously what you see in code and also no redundant part like '1=1' ever appear, also it's easier to debug and optimize as long as you can see the whole picture.
## Installation
### use pip
```bash
$ pip install mysql-cli
```
### use git repository
```text
# this is requirements.txt
# git+https://gitee.com/will4j/mysql-cli-py.git@main#egg=mysql-cli
git+https://github.com/will4j/mysql-cli-py.git@main#egg=mysql-cli
```
```bash
$ pip install -r requirements.txt
```
## Usage
### Basic example
```python
import mysql_cli
from mysql_cli import Select
# init MySQL connection pool
mysql_cli.init_from_conf_file("tests/test_mysql.toml")
@Select("select id, name, cnt from my_test where name = ? limit 1;")
def select_one(name):
return name
assert select_one("hello") == {'id': 1, 'name': 'hello', 'cnt': 2}
```
## Initialization
use `dict` or `toml` configuration file to init MySQL connection. you can find all options in [Python Connection Arguments](https://dev.mysql.com/doc/connector-python/en/connector-python-connectargs.html).
### Init from toml
```toml
# mysql_conf.toml
host = "127.0.0.1"
port = 3306
db = "my_db"
user = "root"
charset = "utf8mb4"
collation = "utf8mb4_general_ci"
pool_name = "my_pool"
pool_size = 5
# Important, for default is not do autocommit
autocommit = true
use_pure = true
```
```python
import mysql_cli
mysql_cli.init_from_conf_file("mysql_conf.toml")
```
### Init from dict
```python
import mysql_cli
mysql_conf = {
"host": "127.0.0.1",
"port": 3306,
"db": "my_db",
"user": "root",
"charset": "utf8mb4",
"collation": "utf8mb4_general_ci",
"pool_name": "my_pool",
"pool_size": 5,
"autocommit": True,
"use_pure": True,
}
mysql_cli.init_from_conf_dict(mysql_conf)
```
## Operate with connection
`mysql_cli.get_connection()`returns a `PooledMySQLConnection`
```python
import mysql_cli
with mysql_cli.get_connection() as cnx:
with cnx.cursor() as cur:
cur.execute("select * from my_test where id = ?", (1,))
print(cur.fetchone())
```
## Query Decorators
All decorators use pooled connections for better performance, also use prepared statements to prevent SQL inject attacks.
You can use ether Format (%s) or qmark (?) in SQL statement.
For named format (like %(field_name)s) is not supported in prepared statement, SQL parameters should be formatted and returned by the decorated method as tuple.
Decorators will take care of the creation and recycling of connections and cursors, as well as transactions, all you need to do is focusing on SQL business logics.
### Insert
`Insert` execute single insert SQL and returns `lastrowid`.
`BatchInsert` execute the insert SQL many times in one transaction and returns `affected_rows`.
```python
from mysql_cli import Insert, BatchInsert
@Insert("insert into my_test (name, cnt) values (?, ?);")
def insert(row: dict):
return row["name"], row["cnt"]
@BatchInsert("insert into my_test (name, cnt) values (?, ?);")
def batch_insert(rows):
return tuple((row["name"], row["cnt"]) for row in rows)
assert insert({"name": "hello", "cnt": 2}) == 1 # lastrowid
params = [{"name": "world", "cnt": 1}, {"name": "world", "cnt": 2}]
assert batch_insert(params) == 2 # affected_rows
```
### Select
`Select` execute select SQL and return one row.
`SelectMany` execute select SQL and return list of rows.
Returned row is dictionary by default, you can set `dictionary=False` to return row as tuple.
```python
from mysql_cli import Select, SelectMany
@Select("select id, name, cnt from my_test where name = ? limit 1;", dictionary=False)
def select_one_return_tuple(name):
return name
@Select("select id, name, cnt from my_test where name = ? limit 1;")
def select_one_return_dict(name):
return name
@SelectMany("select name, cnt from my_test where name = ? order by cnt asc;")
def select_many_by_name(name):
return name
row = select_one_return_tuple("hello")
assert row == (1, 'hello', 2)
row = select_one_return_dict("hello")
assert row == {'id': 1, 'name': 'hello', 'cnt': 2}
rows = select_many_by_name("hello")
assert len(rows) == 2
assert rows[0] == {'id': 1, 'name': 'hello', 'cnt': 2}
```
### Update
`Update` execute update SQL and return affected row number.
```python
from mysql_cli import Update
@Update("update my_test set cnt = ? where name = ? limit ?;")
def update_cnt_by_name(name, cnt, limit=10):
return cnt, name, limit
assert update_cnt_by_name("update_many", 0) == 3 # affected_rows
```
### Delete
`Delete` execute delete SQL and return affected row number.
```python
from mysql_cli import Delete
@Delete("delete from my_test where name = ? limit ?;")
def delete_by_name(name, limit=10):
return name, limit
assert delete_by_name("delete_many") == 2 # affected_rows
```
### Transactional
`Transactional` will start a SQL transaction and control commit or rollback based on decorated method's return. Connection and cursor will be shared in thread local among query decorators. If `Transactional` appears multiple times, they will be merged into one transaction, the very first `Transactional` decides whether to commit or rollback transaction.
```python
from mysql_cli import BatchInsert, Select, Transactional
@BatchInsert("insert into my_test (name, cnt) values (?, ?);")
def batch_insert(params):
return tuple((row["name"], row["cnt"]) for row in params)
@Select("select id, name, cnt from my_test where name = ? limit 1;")
def select_one_return_dict(name):
return name
@Transactional
def transaction_rollback():
params = [{"name": "tx_rollback", "cnt": 1}, {"name": "tx_rollback", "cnt": 2}]
batch_insert(params)
assert select_one_return_dict("tx_rollback")["name"] == "tx_rollback"
raise RuntimeError("rollback")
transaction_rollback()
assert select_one_return_dict("tx_rollback") is None
```
## References
1. https://dev.mysql.com/doc/connector-python/en/connector-python-api-mysqlcursorprepared.html
Raw data
{
"_id": null,
"home_page": "",
"name": "mysql-cli",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": "",
"keywords": "MySQL,Python,client",
"author": "",
"author_email": "William Wang <williamw0825@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/28/5f/eaab7d5dcb99e835ff43de8ac502ddd144aca1c60faae3d86c0668b64512/mysql-cli-1.0.0.tar.gz",
"platform": null,
"description": "# MySQL Python Client\nA MySQL python client based on [mysql-connector-python](https://github.com/mysql/mysql-connector-python), with friendly interface and full-featured client configuration. \nCompare with the way which only takes table names and SQL parameters as input and assembles SQL statement internally, I personally prefer developer taking complete control of SQL, making it flexible which means you can use whatever SQL feature as you need, clear which means the final statement is obviously what you see in code and also no redundant part like '1=1' ever appear, also it's easier to debug and optimize as long as you can see the whole picture.\n\n## Installation\n### use pip\n```bash\n$ pip install mysql-cli\n```\n### use git repository\n```text\n# this is requirements.txt\n# git+https://gitee.com/will4j/mysql-cli-py.git@main#egg=mysql-cli\ngit+https://github.com/will4j/mysql-cli-py.git@main#egg=mysql-cli \n```\n```bash\n$ pip install -r requirements.txt\n```\n\n## Usage\n### Basic example\n```python\nimport mysql_cli\nfrom mysql_cli import Select\n\n\n# init MySQL connection pool\nmysql_cli.init_from_conf_file(\"tests/test_mysql.toml\")\n\n\n@Select(\"select id, name, cnt from my_test where name = ? limit 1;\")\ndef select_one(name):\n return name\n \n \nassert select_one(\"hello\") == {'id': 1, 'name': 'hello', 'cnt': 2}\n```\n\n## Initialization\nuse `dict` or `toml` configuration file to init MySQL connection. you can find all options in [Python Connection Arguments](https://dev.mysql.com/doc/connector-python/en/connector-python-connectargs.html).\n### Init from toml\n\n```toml\n# mysql_conf.toml\nhost = \"127.0.0.1\"\nport = 3306\ndb = \"my_db\"\nuser = \"root\"\ncharset = \"utf8mb4\"\ncollation = \"utf8mb4_general_ci\"\npool_name = \"my_pool\"\npool_size = 5\n# Important, for default is not do autocommit\nautocommit = true\nuse_pure = true\n```\n```python\nimport mysql_cli\n\n\nmysql_cli.init_from_conf_file(\"mysql_conf.toml\")\n```\n### Init from dict\n```python\nimport mysql_cli\n\nmysql_conf = {\n \"host\": \"127.0.0.1\",\n \"port\": 3306,\n \"db\": \"my_db\",\n \"user\": \"root\",\n \"charset\": \"utf8mb4\",\n \"collation\": \"utf8mb4_general_ci\",\n \"pool_name\": \"my_pool\",\n \"pool_size\": 5,\n \"autocommit\": True,\n \"use_pure\": True,\n}\n\nmysql_cli.init_from_conf_dict(mysql_conf)\n```\n\n## Operate with connection\n`mysql_cli.get_connection()`returns a `PooledMySQLConnection`\n```python\nimport mysql_cli\n\n\nwith mysql_cli.get_connection() as cnx:\n with cnx.cursor() as cur:\n cur.execute(\"select * from my_test where id = ?\", (1,))\n print(cur.fetchone())\n```\n\n## Query Decorators\nAll decorators use pooled connections for better performance, also use prepared statements to prevent SQL inject attacks. \nYou can use ether Format (%s) or qmark (?) in SQL statement. \nFor named format (like %(field_name)s) is not supported in prepared statement, SQL parameters should be formatted and returned by the decorated method as tuple. \nDecorators will take care of the creation and recycling of connections and cursors, as well as transactions, all you need to do is focusing on SQL business logics. \n### Insert\n`Insert` execute single insert SQL and returns `lastrowid`. \n`BatchInsert` execute the insert SQL many times in one transaction and returns `affected_rows`.\n\n```python\nfrom mysql_cli import Insert, BatchInsert\n\n\n@Insert(\"insert into my_test (name, cnt) values (?, ?);\")\ndef insert(row: dict):\n return row[\"name\"], row[\"cnt\"]\n\n\n@BatchInsert(\"insert into my_test (name, cnt) values (?, ?);\")\ndef batch_insert(rows):\n return tuple((row[\"name\"], row[\"cnt\"]) for row in rows)\n\n\nassert insert({\"name\": \"hello\", \"cnt\": 2}) == 1 # lastrowid\nparams = [{\"name\": \"world\", \"cnt\": 1}, {\"name\": \"world\", \"cnt\": 2}]\nassert batch_insert(params) == 2 # affected_rows\n```\n### Select\n`Select` execute select SQL and return one row. \n`SelectMany` execute select SQL and return list of rows.\nReturned row is dictionary by default, you can set `dictionary=False` to return row as tuple.\n```python\nfrom mysql_cli import Select, SelectMany\n\n\n@Select(\"select id, name, cnt from my_test where name = ? limit 1;\", dictionary=False)\ndef select_one_return_tuple(name):\n return name\n\n\n@Select(\"select id, name, cnt from my_test where name = ? limit 1;\")\ndef select_one_return_dict(name):\n return name\n\n\n@SelectMany(\"select name, cnt from my_test where name = ? order by cnt asc;\")\ndef select_many_by_name(name):\n return name\n\n\nrow = select_one_return_tuple(\"hello\")\nassert row == (1, 'hello', 2)\nrow = select_one_return_dict(\"hello\")\nassert row == {'id': 1, 'name': 'hello', 'cnt': 2}\nrows = select_many_by_name(\"hello\")\nassert len(rows) == 2\nassert rows[0] == {'id': 1, 'name': 'hello', 'cnt': 2}\n```\n### Update\n`Update` execute update SQL and return affected row number.\n```python\nfrom mysql_cli import Update\n\n\n@Update(\"update my_test set cnt = ? where name = ? limit ?;\")\ndef update_cnt_by_name(name, cnt, limit=10):\n return cnt, name, limit\n\n\nassert update_cnt_by_name(\"update_many\", 0) == 3 # affected_rows\n```\n### Delete\n`Delete` execute delete SQL and return affected row number.\n```python\nfrom mysql_cli import Delete\n\n\n@Delete(\"delete from my_test where name = ? limit ?;\")\ndef delete_by_name(name, limit=10):\n return name, limit\n\n\nassert delete_by_name(\"delete_many\") == 2 # affected_rows\n```\n\n### Transactional\n`Transactional` will start a SQL transaction and control commit or rollback based on decorated method's return. Connection and cursor will be shared in thread local among query decorators. If `Transactional` appears multiple times, they will be merged into one transaction, the very first `Transactional` decides whether to commit or rollback transaction.\n\n```python\nfrom mysql_cli import BatchInsert, Select, Transactional\n\n\n@BatchInsert(\"insert into my_test (name, cnt) values (?, ?);\")\ndef batch_insert(params):\n return tuple((row[\"name\"], row[\"cnt\"]) for row in params)\n\n\n@Select(\"select id, name, cnt from my_test where name = ? limit 1;\")\ndef select_one_return_dict(name):\n return name\n\n\n@Transactional\ndef transaction_rollback():\n params = [{\"name\": \"tx_rollback\", \"cnt\": 1}, {\"name\": \"tx_rollback\", \"cnt\": 2}]\n batch_insert(params)\n\n assert select_one_return_dict(\"tx_rollback\")[\"name\"] == \"tx_rollback\"\n raise RuntimeError(\"rollback\")\n\n\ntransaction_rollback()\nassert select_one_return_dict(\"tx_rollback\") is None\n```\n\n## References\n1. https://dev.mysql.com/doc/connector-python/en/connector-python-api-mysqlcursorprepared.html\n",
"bugtrack_url": null,
"license": "",
"summary": "A MySQL Python Client",
"version": "1.0.0",
"split_keywords": [
"mysql",
"python",
"client"
],
"urls": [
{
"comment_text": "",
"digests": {
"md5": "a8460151c4b999a4d5b1ef1fe98072ec",
"sha256": "ddd5d202510d9179c7c4a7321be4e13480410ff8681c877a67b8a8cfdc99260e"
},
"downloads": -1,
"filename": "mysql_cli-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a8460151c4b999a4d5b1ef1fe98072ec",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 7160,
"upload_time": "2022-12-11T15:48:57",
"upload_time_iso_8601": "2022-12-11T15:48:57.946518Z",
"url": "https://files.pythonhosted.org/packages/e4/18/5c291d66571a0a3eeb3fdb2f82e31f562e6e2a12deeb83819ea4c5390be5/mysql_cli-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"md5": "867f4803d1f45dba6cde92510015b030",
"sha256": "fe6d3270a661430fc360baa0b89e5156d509ce994ed852de84fee8780514bd53"
},
"downloads": -1,
"filename": "mysql-cli-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "867f4803d1f45dba6cde92510015b030",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 6848,
"upload_time": "2022-12-11T15:48:59",
"upload_time_iso_8601": "2022-12-11T15:48:59.643084Z",
"url": "https://files.pythonhosted.org/packages/28/5f/eaab7d5dcb99e835ff43de8ac502ddd144aca1c60faae3d86c0668b64512/mysql-cli-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2022-12-11 15:48:59",
"github": false,
"gitlab": false,
"bitbucket": false,
"lcname": "mysql-cli"
}
