一、安装与配置postgreSQL
参考文章：
在Ubuntu上安装与简单使用PostgreSQL数据库
ubuntu防火墙设置 （ufw、iptables）
PostgreSQL 12.2 手册
1. 安装postgresql与postgresql-contrib
$sudo apt install postgresql
$sudo apt install postgesql-contrib
# 安装完成后查看服务是否正常运行
$service postgresql status
# 查看所运行的端口
$cat /etc/services |grep postgresql
2. 安装完成后会创建一个拥有所有权限的特殊用户postgres，需要使用这个用户才可以登录数据库命令行
$sudo su postgres
$psql
# 或$psql --username postgres
出现下面提示说明登录成功
psql (14.11 (Ubuntu 14.11-0ubuntu0.22.04.1))
Type "help" for help.
# 命令提示符的postgres表示当前所在的数据库是postgres
postgres=#
psql命令行模式下，可以执行sql语句和元命令。
元命令是指以\开头的命令，是postgreSQL特有的用于管理数据库的命令，元命令末尾可以不加分号，但sql语句需要。
# 查看数据库列表
postgres=# \l
# 查看用户列表
postgres=# \du
# 查看版本 也可以使用 $psql --version
postgres=# select version();
# 查看当前用户
postgres=# select current_user;
# 如果没有加分号命令提示符会变成postgres-#
# postgres=# select version()
# postgres-#
3. 创建postgre用户并授予超级用户权限
超级用户可以绕过所有权限检查，所以不需要再额外分配其他权限。
<span style="color:red">注意：这里创建的是postgre用户，而非linux用户</span>
这里假设我们创建的用户名和密码都是admin
postgres=# CREATE USER admin with PASSWORD 'admin';
CREATE ROLE
postgres=# ALTER USER admin WITH SUPERUSER;
ALTER ROLE
# 等价于
# postgres=# CREATE USER admin WITH PASSWORD 'admin' SUPERUSER;
# 如果只想分配部分权限
# postgres=# CREATE USER guess WITH PASSWORD 'guess' CREATEDB CREATEROLE;
# 查看创建后的角色
postgres=# \du admin
关于角色权限部分更详细的介绍可以查阅官方文档http://www.postgres.cn/docs/12/user-manag.html
4. 退出并以新的用户登录
登录时需要指定数据库名，否则会默认登录到同名数据库（admin），此时该数据库还没有被创建，将会报错。（当然你也可以先使用postgres用户创建该数据库）
postgres=# exit
$psql --username=admin --dbname=postgres
# 或
# $psql -U admin -d postgres
这时候可能会出现如下报错：
FATAL: Peer authentication failed for user "admin"
这是因为psql对本地登录的默认认证方式为peer。peer认证简单来说就是检测当前操纵系统用户名与要登录的数据库用户名是否一致。
你可以直接修改当前系统用户名或新建一个用户来通过peer认证，或者修改认证方式。
5. 修改认证方式
# 命令中的14更换为自己版本号
$sudo nano /etc/postgresql/14/main/pg_hba.conf
找到如下信息：
# Database administrative login by Unix domain socket
local all postgres peer
# TYPE DATABASE USER ADDRESS METHOD
# "local" is for Unix domain socket connections only
local all all peer
# IPv4 local connections:
host all all 127.0.0.1/32 scram-sha-256
# IPv6 local connections:
host all all ::1/128 scram-sha-256
将
# "local" is for Unix domain socket connections only
local all all peer
改为
# "local" is for Unix domain socket connections only
local all all md5
或在上面插入一条记录
# "local" is for Unix domain socket connections only
local all admin md5
local all all peer
然后重启服务
$sudo service postgresql restart
这个文件之后还会再修改
pg_hba.conf(PostgreSQL Host-Based Authentication)是psql默认的客户端认证配置文件，没有被#注释的行被称为一条记录。当psql服务器进行认证时，会从上往下寻找第一条匹配的记录进行认证。
每条记录包含7个域（列），其中带*号的为可选项
域 说明 TYPE 连接类型。 DATABASE 数据库名称，除了具体的数据库名称，还可取值all、sameuser或者@数据库文件名等 USER postgre数据库用户名 ADDRESS* ip地址、域名或一些关键字。当类型不是local时此项必选 IP-MASK* 子网掩码。当给定的ip地址没有指定前缀长度时必选 METHOD 认证方法。 OPTIONS* 认证选项 认证方法主要包括：
trust：无条件允许连接
reject：无条件拒绝连接
scram-sha-256/md5/password：这三种都是基于密码的认证，区别在于密码的存储方式与传输方式，其中password是以明文传输与存储，应当避免使用
peer：检查操作系统用户名是否与数据库用户名匹配，只对本地连接可用。
还有一些基于证书或服务器的认证方式，感兴趣的可以查看官方文档http://postgres.cn/docs/12/auth-pg-hba-conf.html
这里我们暂时只关注前两条记录
# Database administrative login by Unix domain socket local all postgres peer # TYPE DATABASE USER ADDRESS METHOD # "local" is for Unix domain socket connections only local all all peer第一条记录的意思是：在本地使用postgres用户登录所有数据库时，使用peer认证。
第二条记录的意思是：在本地使用所有用户登录所有数据库时，使用peer认证。
当我们在本地使用postgres用户登录时，会匹配第一条记录，然后不再往下匹配；在本地使用其他用户登录时（如刚才注册的admin用户），会匹配第二条记录，于是会进行peer认证。
如果我们不希望对admin进行peer认证，可以为admin添加一条记录，将认证方式设置为基于密码的认证或者trust；或者直接修改所有本地用户的认证方式。
此时应该就可以成功登录了，登录后创建一个同名数据库用以测试
$psql --username=admin --dbname=postgres
postgres=# create database admin;
二、配置远程连接
1. 修改远程连接认证规则
还是打开pg_hba.conf
# 命令中的14更换为自己版本号
$sudo nano /etc/postgresql/14/main/pg_hba.conf
找到
# IPv4 local connections:
host all all 127.0.0.1/32 scram-sha-256
在下面添加一条记录
# IPv4 local connections:
host all all 127.0.0.1/32 scram-sha-256
host all all 0.0.0.0/0 md5
2. 配置监听端口
打开postgresql.conf
# 命令中的14更换为自己版本号
$sudo nano /etc/postgresql/14/main/postgresql.conf
找到
# - Connection Settings -
#listen_addresses = 'localhost' # what IP address(es) to listen on;
# comma-separated list of addresses;
# defaults to 'localhost'; use '*' for all
# (change requires restart)
port = 5432 # (change requires restart)
max_connections = 100 # (change requires restart)
将监听地址从localhost改为*
# - Connection Settings -
#listen_addresses = 'localhost' # what IP address(es) to listen on;
# comma-separated list of addresses;
# defaults to 'localhost'; use '*' for all
# (change requires restart)
listen_addresses = '*'
port = 5432 # (change requires restart)
max_connections = 100 # (change requires restart)
重启服务
$sudo service postgresql restart
3. 修改防火墙，放行端口
# 安装ufw
$sudo apt install ufw
# 启动防火墙并设置开机启动
$sudo ufw enable
$sudo ufw default deny
# 放行5432端口
$sudo ufw allow 5432/tcp
# 或
# $sudo ufw allow postgresql
4. 测试连接
查看虚拟机内网ip，我这里是192.168.65.130
$ip addr show |grep -w inet
使用pycharm database测试数据库连接
db连接测试.png
三、使用FastAPI连接postgreSQL数据库
下面内容基于FastAPI官方文档https://fastapi.tiangolo.com/zh/tutorial/sql-databases/
主要步骤：
- 安装驱动并配置数据库连接
- 创建SQLAlchemy ORM模型（DAO，用于持久层，在这里也被称为model）
- 创建pydantic模型（DTO，用于服务层与展示层的传输，在这里也被称为schema）
- 编写CRUD工具函数
- 编写依赖，并注入到路径函数中
之所以要为一个实体创建不同的模型（类），除了因为不同的层面对实体的关注点不同，所需字段的数据类型不同之外，更重要的是FastAPI只能从请求中读取或从响应中返回pydantic模型对象，SQLAlchemy只能从数据库读取或往数据库写入SQLAlchemy ORM模型对象。
数据库连接模型.png
这个项目的文件目录结构如下：
.
└── sql_app
├── __init__.py
├── crud.py
├── database.py
├── main.py
├── models.py
└── schemas.py
安装依赖
# 数据库连接驱动
pip install psycopg2
# ORM框架
pip install sqlalchemy
1. 在database.py中，配置数据库连接参数。
主要是需要调用sessionmaker()创建session工厂类以及调用declarative_base()创建model基类
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
username = "admin"
password = "admin"
host = "192.168.65.130"
port = 5432
database = "admin"
DATABASE_URL = f"postgresql://{username}:{password}@{host}:{port}/{database}"
# 数据库引擎
engine = create_engine(url=DATABASE_URL)
# Seesion工厂类，后续通过session = SessionLocal()来获取一个数据库连接会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# ORM模型基类，所有ORM模型都需要继承这个基类
Base = declarative_base()
值得一提的是，SQLAlchemy 2.0更推荐使用继承的方式声明Base基类
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
2. 在models.py中创建ORM模型
ORM模型对应的表不一定需要存在，稍后可以根据模型创建该表。
这里给出基于SQLAlchemy 2.0的代码，主要区别在于使用mapped_column代替Column，这样做的好处是可以声明属性的类型，从而获得IDE的代码提示和类型检查
# SQLAlchemy 1.x
id = Column(Integer, primary_key=True)
# SQLAlchemy 2.x
id: Mapped[int] = mapped_column(primary_key=True)
from typing import List
from sqlalchemy import ForeignKey
from sqlalchemy.orm import mapped_column, Mapped, relationship
from database import Base
# 继承Base类
class User(Base):
# 声明对应的表名
__tablename__ = 'users'
# 属性
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(unique=True)
hashed_password: Mapped[str] = mapped_column()
is_active: Mapped[bool] = mapped_column(default=True)
# 多表连接关系
# back_populates="owner"表示与Item类的onwer表形成双向连接
items: Mapped[List["Item"]] = relationship(back_populates="owner")
class Item(Base):
__tablename__ = 'items'
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column()
description: Mapped[str] = mapped_column()
owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
owner: Mapped[User] = relationship(back_populates="items")
3. 在schemas.py中创建pydantic模型
from pydantic import BaseModel
# 将共有属性抽象成基类
class ItemBase(BaseModel):
title: str
description: str | None = None
# 用于请求体接受数据
class ItemCreate(ItemBase):
pass
# 用于响应体返回数据
class Item(ItemBase):
id: int
owner_id: int
# 创建Config子类并设置from_attributes=True
# 这是在告诉这个pydantic模型，在从一个东西读取数据时，除了尝试把这个东西看做字典，还应当把他看做对象
# 即需要尝试self.id = data['id']与self.id = data.id
class Config:
# orm_mode = True # pydantic v1写法
from_attributes = True # pydantic v2写法
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
is_active: bool
items: list[Item] = []
class Config:
# orm_mode = True
from_attributes = True
4. 在crud.py中编写CRUD工具函数
from sqlalchemy.orm import Session
import models
import schemas
def get_user(db: Session, user_id: int):
# 这种写法会报黄
# return db.query(models.User).filter(models.User.id == user_id).first()
# 这种写法不会
return db.query(models.User).filter_by(id=user_id).first()
def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter_by(email=email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()
def get_item(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Item).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
fake_hashed_password = user.password + "hashed"
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
db_item = models.Item(**item.dict(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
5. 在main.app编写接口
在main.py中，有几个地方值得关注
创建所有ORM对应的数据表：
所有继承自Base的子类都共享一个metadata对象，每个子类所对应表的表结构和列信息都会注册到该metadata对象中，使用metadata.create_all()即可在数据库中创建这些表。
models.Base.metadata.create_all(bind=engine)
定义用于获取数据库连接会话的依赖项(Dependency)：
# Dependency
def get_db():
# 使用SesionLocal()创建一个临时会话
db = SessionLocal()
try:
yield db
except HTTPException:
raise
except Exception as e:
print(e.args)
finally:
db.close()
在参数中引入该依赖项：
def create_user(user: schemas.UserCreate = Body(), db: Session = Depends(get_db)):
指定返回模型response_model:
由于从数据库中读取的数据对象是ORM模型（models.User)，不能直接返回，需要先转换成pydantic模型（schemas.User）。我们可以手动转换，但其实只需要在路径装饰器上指定response_model参数，FastAPI就会自动帮我们转换
@app.post("/users/", response_model=schemas.User)
当然该pydantic模型必须先配置from_attributes = True
# schemas.py
class User(UserBase):
id: int
is_active: bool
items: list[Item] = []
class Config:
# orm_mode = True
from_attributes = True
完整代码如下
from fastapi import FastAPI, Depends, HTTPException, Body, status, Path
from sqlalchemy.orm import Session
import crud
import models
import schemas
from database import SessionLocal, engine
app = FastAPI()
models.Base.metadata.create_all(bind=engine)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
except HTTPException:
raise
except Exception as e:
print(e.args)
finally:
db.close()
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate = Body(), db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, user.email)
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
return crud.create_user(db, user)
@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
db_users = crud.get_users(db, skip, limit)
return db_users
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int = Path(), db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id)
if not db_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return db_user
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
user_id: int = Path(),
item: schemas.ItemCreate = Body(),
db: Session = Depends(get_db)
):
return crud.create_user_item(db, item, user_id)
@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_item(db, skip, limit)
return items
6. 启动uvicorn服务器， 访问http://127.0.0.1:8000/docs进行测试
cd sql_app
uvicorn main:app --reload
