파이썬 조작 SQL-파이썬 튜토리얼-php.cn

집

백엔드 개발

파이썬 튜토리얼

파이썬 조작 SQL

高洛峰

Dec 02, 2016 pm 05:10 PM

python

pymsql은 Python에서 MySQL을 운영하기 위한 모듈입니다. 사용법은 MySQLdb와 거의 같습니다

pip3 install pymysql

을 다운로드하여 설치하세요. 2. 연산 활용

1. SQL 실행

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
# 创建连接
conn = pymysql.connect(host=&#39;127.0.0.1&#39;, port=3306, user=&#39;root&#39;, passwd=&#39;123&#39;, db=&#39;t1&#39;)
# 创建游标
cursor = conn.cursor()
  
# 执行SQL，并返回收影响行数
effect_row = cursor.execute("update hosts set host = &#39;1.1.1.2&#39;")
  
# 执行SQL，并返回受影响行数
#effect_row = cursor.execute("update hosts set host = &#39;1.1.1.2&#39; where nid > %s", (1,))
  
# 执行SQL，并返回受影响行数
#effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
  
  
# 提交，不然无法保存新建或者修改的数据
conn.commit()
  
# 关闭游标
cursor.close()
# 关闭连接
conn.close()

로그인 후 복사

2. 자동 증가 ID로 새로 생성된 데이터 가져오기

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
conn = pymysql.connect(host=&#39;127.0.0.1&#39;, port=3306, user=&#39;root&#39;, passwd=&#39;123&#39;, db=&#39;t1&#39;)
cursor = conn.cursor()
cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
conn.commit()
cursor.close()
conn.close()
  
# 获取最新自增ID
new_id = cursor.lastrowid

로그인 후 복사

3. >

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
conn = pymysql.connect(host=&#39;127.0.0.1&#39;, port=3306, user=&#39;root&#39;, passwd=&#39;123&#39;, db=&#39;t1&#39;)
cursor = conn.cursor()
cursor.execute("select * from hosts")
  
# 获取第一行数据
row_1 = cursor.fetchone()
  
# 获取前n行数据
# row_2 = cursor.fetchmany(3)
# 获取所有数据
# row_3 = cursor.fetchall()
  
conn.commit()
cursor.close()
conn.close()

로그인 후 복사

참고: 데이터를 가져올 때 다음과 같이cursor.scroll(num, mode)를 사용하여 커서 위치를 이동할 수 있습니다.

cursor.scroll(1, mode ='relative') # 현재 위치 이동에 대한

cursor.scroll(2,mode='absolute') # 상대 절대 위치 이동

4. 기본값입니다. 얻은 데이터는 조상 유형입니다. 즉,

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
conn = pymysql.connect(host=&#39;127.0.0.1&#39;, port=3306, user=&#39;root&#39;, passwd=&#39;123&#39;, db=&#39;t1&#39;)
  
# 游标设置为字典类型
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
  
result = cursor.fetchone()
  
conn.commit()
cursor.close()
conn.close()

로그인 후 복사

SQLAlchemy

SQLAlchemy는 Python 프로그래밍 언어의 ORM 프레임워크입니다. 데이터베이스 API를 사용하고 데이터베이스 작업을 위해 관계형 개체 매핑을 사용합니다. 즉, 개체를 SQL로 변환한 다음 데이터 API를 사용하여 SQL을 실행하고 실행 결과를 얻습니다.

설치:

pip3 install SQLAlchemy

파이썬 조작 SQL

SQLAlchemy 자체는 데이터베이스를 작동할 수 없으며 타사 플러그에 의존해야 합니다. -pymsql과 같은 Dialect는 데이터 API와 통신하고 다양한 구성 파일에 따라 다양한 데이터베이스 API를 호출하여 다음과 같은 데이터베이스 작업을 수행하는 데 사용됩니다.

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

로그인 후 복사

1. 내부 처리

엔진 /ConnectionPooling/Dialect를 사용하여 데이터베이스 작업을 수행합니다. 엔진은 ConnectionPooling을 사용하여 데이터베이스에 연결한 후 Dialect를 통해 SQL 문을 실행합니다.

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
  
  
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (&#39;1.1.1.22&#39;, 3)"
# )
  
# 新插入行自增ID
# cur.lastrowid
  
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[(&#39;1.1.1.22&#39;, 3),(&#39;1.1.1.221&#39;, 3),]
# )
  
  
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host=&#39;1.1.1.99&#39;, color_id=3
# )
  
# 执行SQL
# cur = engine.execute(&#39;select * from hosts&#39;)
# 获取第一行数据
# cur.fetchone()
# 获取第n行数据
# cur.fetchmany(3)
# 获取所有数据
# cur.fetchall()

로그인 후 복사

2. ORM 함수 사용

ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect의 모든 구성 요소를 사용하여 데이터 작업을 수행합니다. 클래스를 기반으로 객체를 생성하고, 객체를 SQL로 변환하고, SQL을 실행합니다.

1. 테이블 생성

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
 
Base = declarative_base()
 
# 创建单表
class Users(Base):
    __tablename__ = &#39;users&#39;
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))
 
    __table_args__ = (
    UniqueConstraint(&#39;id&#39;, &#39;name&#39;, name=&#39;uix_id_name&#39;),
        Index(&#39;ix_id_name&#39;, &#39;name&#39;, &#39;extra&#39;),
    )
 
 
# 一对多
class Favor(Base):
    __tablename__ = &#39;favor&#39;
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default=&#39;red&#39;, unique=True)
 
 
class Person(Base):
    __tablename__ = &#39;person&#39;
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
 
 
# 多对多
class Group(Base):
    __tablename__ = &#39;group&#39;
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
 
 
class Server(Base):
    __tablename__ = &#39;server&#39;
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
 
 
class ServerToGroup(Base):
    __tablename__ = &#39;servertogroup&#39;
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey(&#39;server.id&#39;))
    group_id = Column(Integer, ForeignKey(&#39;group.id&#39;))
 
 
def init_db():
    Base.metadata.create_all(engine)
 
 
def drop_db():
    Base.metadata.drop_all(engine)
注：设置外检的另一种方式 ForeignKeyConstraint([&#39;other_id&#39;], [&#39;othertable.other_id&#39;])

로그인 후 복사

2. 연산 테이블

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)

Base = declarative_base()

# 创建单表
class Users(Base):
    __tablename__ = &#39;users&#39;
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

    __table_args__ = (
    UniqueConstraint(&#39;id&#39;, &#39;name&#39;, name=&#39;uix_id_name&#39;),
        Index(&#39;ix_id_name&#39;, &#39;name&#39;, &#39;extra&#39;),
    )

    def __repr__(self):
        return "%s-%s" %(self.id, self.name)

# 一对多
class Favor(Base):
    __tablename__ = &#39;favor&#39;
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default=&#39;red&#39;, unique=True)

    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

class Person(Base):
    __tablename__ = &#39;person&#39;
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref=&#39;pers&#39;)

# 多对多
class ServerToGroup(Base):
    __tablename__ = &#39;servertogroup&#39;
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey(&#39;server.id&#39;))
    group_id = Column(Integer, ForeignKey(&#39;group.id&#39;))
    group = relationship("Group", backref=&#39;s2g&#39;)
    server = relationship("Server", backref=&#39;s2g&#39;)

class Group(Base):
    __tablename__ = &#39;group&#39;
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
    # group = relationship(&#39;Group&#39;,secondary=ServerToGroup,backref=&#39;host_list&#39;)


class Server(Base):
    __tablename__ = &#39;server&#39;

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)




def init_db():
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)


Session = sessionmaker(bind=engine)
session = Session()

로그인 후 복사

obj = Users(name="alex0", extra=&#39;sb&#39;)
session.add(obj)
session.add_all([
    Users(name="alex1", extra=&#39;sb&#39;),
    Users(name="alex2", extra=&#39;sb&#39;),
])
session.commit()

로그인 후 복사

 session.query(Users).filter(Users.id > 2).delete()
 session.commit()

로그인 후 복사

session.query(Users).filter(Users.id > 2).update({"name" : "099"})
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
session.commit()

로그인 후 복사

 ret = session.query(Users).all()
 ret = session.query(Users.name, Users.extra).all()
 ret = session.query(Users).filter_by(name=&#39;alex&#39;).all()
 ret = session.query(Users).filter_by(name=&#39;alex&#39;).first()

로그인 후 복사

ret = session.query(Users).filter_by(name=&#39;alex&#39;).all()
ret = session.query(Users).filter(Users.id > 1, Users.name == &#39;eric&#39;).all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == &#39;eric&#39;).all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=&#39;eric&#39;))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == &#39;eric&#39;)).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == &#39;eric&#39;)).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == &#39;eric&#39;, Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like(&#39;e%&#39;)).all()
ret = session.query(Users).filter(~Users.name.like(&#39;e%&#39;)).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()

로그인 후 복사

본 웹사이트의 성명

본 글의 내용은 네티즌들의 자발적인 기여로 작성되었으며, 저작권은 원저작자에게 있습니다. 본 사이트는 이에 상응하는 법적 책임을 지지 않습니다. 표절이나 침해가 의심되는 콘텐츠를 발견한 경우 admin@php.cn으로 문의하세요.

핫 AI 도구

뜨거운 도구

뜨거운 주제

Gmail 이메일의 로그인 입구는 어디에 있나요?

7515

Cakephp 튜토리얼

1378

Steam의 계정 이름 형식은 무엇입니까?

Win11 활성화 키 영구

NYT 연결 힌트와 답변

Related knowledge

2 시간의 파이썬 계획 : 현실적인 접근 Apr 11, 2025 am 12:04 AM

2 시간 이내에 Python의 기본 프로그래밍 개념과 기술을 배울 수 있습니다. 1. 변수 및 데이터 유형을 배우기, 2. 마스터 제어 흐름 (조건부 명세서 및 루프), 3. 기능의 정의 및 사용을 이해하십시오. 4. 간단한 예제 및 코드 스 니펫을 통해 Python 프로그래밍을 신속하게 시작하십시오.

Redis 대기열을 읽는 방법 Apr 10, 2025 pm 10:12 PM

Redis의 대기열을 읽으려면 대기열 이름을 얻고 LPOP 명령을 사용하여 요소를 읽고 빈 큐를 처리해야합니다. 특정 단계는 다음과 같습니다. 대기열 이름 가져 오기 : "큐 :"와 같은 "대기열 : my-queue"의 접두사로 이름을 지정하십시오. LPOP 명령을 사용하십시오. 빈 대기열 처리 : 대기열이 비어 있으면 LPOP이 NIL을 반환하고 요소를 읽기 전에 대기열이 존재하는지 확인할 수 있습니다.

Redis의 서버 버전을 보는 방법 Apr 10, 2025 pm 01:27 PM

질문 : Redis 서버 버전을 보는 방법은 무엇입니까? 명령 줄 도구 Redis-Cli를 사용하여 연결된 서버의 버전을보십시오. 정보 서버 명령을 사용하여 서버의 내부 버전을보고 정보를 구문 분석하고 반환해야합니다. 클러스터 환경에서 각 노드의 버전 일관성을 확인하고 스크립트를 사용하여 자동으로 확인할 수 있습니다. 스크립트를 사용하여 Python 스크립트와 연결 및 인쇄 버전 정보와 같은보기 버전을 자동화하십시오.

Redis로 서버를 시작하는 방법 Apr 10, 2025 pm 08:12 PM

Redis 서버를 시작하는 단계에는 다음이 포함됩니다. 운영 체제에 따라 Redis 설치. Redis-Server (Linux/MacOS) 또는 Redis-Server.exe (Windows)를 통해 Redis 서비스를 시작하십시오. Redis-Cli Ping (Linux/MacOS) 또는 Redis-Cli.exe Ping (Windows) 명령을 사용하여 서비스 상태를 확인하십시오. Redis-Cli, Python 또는 Node.js와 같은 Redis 클라이언트를 사용하여 서버에 액세스하십시오.

비즈니스 요구에 따라 Redis 메모리 크기를 설정하는 방법은 무엇입니까? Apr 10, 2025 pm 02:18 PM

Redis 메모리 크기 설정은 다음 요소를 고려해야합니다. 데이터 볼륨 및 성장 추세 : 저장된 데이터의 크기 및 성장 속도를 추정하십시오. 데이터 유형 : 다른 유형 (예 : 목록, 해시)은 다른 메모리를 차지합니다. 캐싱 정책 : 전체 캐시, 부분 캐시 및 단계 정책은 메모리 사용에 영향을 미칩니다. 비즈니스 피크 : 트래픽 피크를 처리하기에 충분한 메모리를 남겨 두십시오.

Redis 메모리 구성 매개 변수는 무엇입니까? Apr 10, 2025 pm 02:03 PM

** Redis 메모리 구성의 핵심 매개 변수는 MaxMemory로 Redis가 사용할 수있는 메모리의 양을 제한합니다. 이 한계가 초과 될 때, Redis는 Maxmemory-Policy에 따라 제거 전략을 실행합니다. 다른 관련 매개 변수로는 MaxMemory-Samples (LRU 샘플 수량), RDB- 압축이 있습니다

메모리에 대한 Redis 지속성의 영향은 무엇입니까? Apr 10, 2025 pm 02:15 PM

Redis Persistence는 추가 메모리를 차지하고 RDB는 스냅 샷을 생성 할 때 메모리 사용량을 일시적으로 증가시키고 AOF는 로그를 추가 할 때 계속 메모리를 차지합니다. 영향 요인에는 데이터 볼륨, 지속성 정책 및 Redis 구성이 포함됩니다. 영향을 완화하려면 RDB 스냅 샷 정책을 합리적으로 구성하고 구성 최적화, 하드웨어 업그레이드 및 메모리 사용을 모니터링 할 수 있습니다. 또한 성능과 데이터 보안 사이의 균형을 찾는 것이 중요합니다.