s = jenkins.Jenkins(url=URL, username=USERNAME, password=PASSWORD)

jobs = s.get_jobs(view_name=VIEW_NAME)

for job in jobs:
    name = job['name']
    config = s.get_job_config(name)

    content = f'''</permission>
    <permission>hudson.model.Item.Build:{USER}</permission>
    <permission>hudson.model.Item.Cancel:{USER}</permission>
    <permission>hudson.model.Item.Configure:{USER}</permission>
    <permission>hudson.model.Item.Delete:{USER}</permission>
    <permission>hudson.model.Item.Move:{USER}</permission>
    <permission>hudson.model.Item.Read:{USER}</permission>
    <permission>hudson.model.Item.Workspace:{USER}</permission>
    '''

    new = config.replace('</permission>', content, 1)

    s.reconfig_job(name, new)

session = HTMLSession()

# Get crsf token
r = session.get(url=f'{URL}/user/login')
csrf = r.html.xpath('//input[@name="_csrf"]/@value', first=True)

# Login
data = {
    '_csrf': csrf,
    'user_name': USERNAME,
    'password': PASSWORD,
    'login_source': 2
}
session.post(url=f'{URL}/user/login', data=data)

# Get login crsf token
r = session.get(url=f'{URL}')
login_csrf = r.html.xpath('//meta[@name="_csrf"]/@content', first=True)

# Set tag for project from csv file
projects = get_test_csv_data('build.csv')
for project in projects:
    if project[1] == 'Y':

        build_data = {
            'tag_name': project[2],
            'tag_target': project[3],
            'title': project[4],
            'content': project[5],
            '_csrf': login_csrf
        }

        print(f'===== set tag {project[2]} for {project[0]} =====')
        print(build_data)
        r = session.post(url=f'{URL}/{OWNER}/{project[0]}/releases/new', data=build_data)

        # Verify set correctly
        try:
            res = session.get(f'{URL}/{OWNER}/{project[0]}/releases')
            version = res.html.xpath('//ul[@id="release-list"]//a', first=True).text
            if version == project[2]:
                print(f'===== set tag {project[2]} for {project[0]} successfully =====')
        except:
            print(f'===== set tag {project[2]} for {project[0]} failed =====')
        print()

session = HTMLSession()

# Get crsf token
r = session.get(url=URL)
csrf = r.html.xpath('//input[@name="_csrf"]', first=True)

# Login
data = {
    '_csrf': csrf.attrs.get('value'),
    'user_name': USERNAME,
    'password': PASSWORD,
    'login_source': 2
}
session.post(url=URL, data=data)

# Get project latest release version
release_version = {}
for p in PROJECT_LIST:
    res = session.get(f'https://{URL}/{p}/releases')

    version = res.html.xpath('//ul[@id="release-list"]//a', first=True).text

    release_version[p] = version

print(json.dumps(release_version, indent=4))

con = Confluence(url=URL, username=USERNAME, password=PASSWORD)

body = '<div class="table-wrap"><table class="relative-table wrapped confluenceTable">版本信息</table></div>'
con.create_page(space='BDP', title='testtesttest', parent_id=85295571,
                body=body)

from jose import jwt
from datetime import datetime, timedelta

# 加密密钥
SECRET_KEY = "testjwt"

# 设置过期时间 示例5分钟
expire = datetime.utcnow() + timedelta(minutes=5)

# exp 是固定写法必须得传  username和uid是自己存的值
to_encode = {"exp": expire, "username": "admin", "uid": "12345"}

# 生成token
token = jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")

from jose.exceptions import ExpiredSignatureError, JWTError
try:
    payload = jwt.decode(token, SECRET_KEY, algorithms="HS256" )
    print(payload)
except ExpiredSignatureError as e:
    print("token过期")
except JWTError as e:
    print("token验证失败")

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from models import crud,schemas
from dependencies import get_db

router = APIRouter()


# 新建用户
@router.post("/users/", response_model=schemas.User, tags=['users'])
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db, user=user)


# 通过id查询用户
@router.get("/user/{user_id}", response_model=schemas.User, tags=['users'])
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


# 所有用户
@router.get('/users/', response_model=List[schemas.User], tags=['users'])
def read_users(start: int=0, limit: int=10, db: Session = Depends(get_db)):
    return crud.get_users(db, start, limit)

# Delete user
@router.delete('/user/{user_id}', response_model=schemas.User, tags=['users'])
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.delete_user(db, user_id=user_id)
    if not db_user:
        raise HTTPException(status_code=400, detail='User not found')
    return db_user

# Update User
@router.put('/user/{user_id}', response_model=schemas.User, tags=['users'])
def update_user(user_id: int, update_user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.update_user(db, user_id=user_id, update_user=update_user)
    if not db_user:
        raise HTTPException(status_code=400, detail='User not found')
    return db_user

# 导入 FastAPI
from fastapi import FastAPI
from routers import users
import uvicorn
import config

# 实例化 FastAPI
app = FastAPI()

app.include_router(users.router)

├─db
│      crud.py
│      database.py
│      models.py
│      schemas.py
│      __init__.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@127.0.0.1:3306/test"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, encoding='utf8', echo=True
)

# SQLAlchemy中，CRUD是通过会话进行管理的，所以需要先创建会话，
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基本映射类
Base = declarative_base()

from sqlalchemy import Boolean, Column, Integer, String
from database import Base


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(32), unique=True, index=True)
    hash_password = Column(String(32))
    is_active = Column(Boolean, default=True)

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    """
    请求模型验证：
    email:
    password:
    """
    password: str

class UserUpdate(UserBase):
    is_active: bool

class User(UserBase):
    """
    响应模型：
    id:
    email:
    is_active
    并且设置orm_mode与之兼容
    """
    id: int
    is_active: bool

    class Config:
        orm_mode = True

from sqlalchemy.orm import Session
import models, schemas


# Get user by id
def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

# Get all users
def get_users(db: Session, start: int=0, limit: int=10):
    return db.query(models.User).offset(start).limit(limit).all()

# creae user
def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    user = models.User(email=user.email, hash_password=fake_hashed_password)
    db.add(user)
    db.commit()  # 提交保存到数据库中
    db.refresh(user)  # 刷新
    return user

# Delete user
def delete_user(db: Session, user_id: int):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
        db.flush()
        return user


# Update user
def update_user(db: Session, user_id: int, update_user: schemas.UserUpdate):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user:
        update_user = update_user.dict(exclude_unset=True)
        for k, v in update_user.items():
            setattr(user, k, v)
        db.commit()
        db.flush()
        db.refresh(user)
        return user

Base.metadata.create_all(bind=engine) #数据库初始化，如果没有库或者表，会自动创建

# 实例化 FastAPI
app = FastAPI()

# Dependency
def get_db():
    """
    每一个请求处理完毕后会关闭当前连接，不同的请求使用不同的连接
    :return:
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 新建用户
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.db_create_user(db=db, user=user)


# 通过id查询用户
@app.get("/user/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# 所有用户
@app.get('/users/', response_model=List[schemas.User])
def read_users(start: int=0, limit: int=10, db: Session = Depends(get_db)):
    return crud.get_users(db, start, limit)

# Delete user
@app.delete('/user/{user_id}', response_model=schemas.User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.delete_user(db, user_id=user_id)
    if not db_user:
        raise HTTPException(status_code=400, detail='User not found')
    return db_user

# Update User
@app.put('/user/{user_id}', response_model=schemas.User)
def update_user(user_id: int, update_user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.update_user(db, user_id=user_id, update_user=update_user)
    if not db_user:
        raise HTTPException(status_code=400, detail='User not found')
    return db_user

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get('/users/')
def users(token: str = Depends(oauth2_scheme)):
    return {'token': token}

fake_user = {
    'admin':{
        'username': 'admin',
        'hash_password': 'fakehashadmin',
        'enabled': True
    }
}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")# 这个参数的作用是当输入完用户密码后, 点击Authorize按钮后, 会请求127.0.0.1:8000/token这个网址, 把用户密码通过Form表单的形式传递给这个请求, 即这个/token就是我们网站的后台登录接口.


def fake_hash_password(password: str):
    return "fakehash" + password

class User(BaseModel):
    username: str
    enbaled: Optional[bool] = None

class UserPwd(User):
    hash_password: str


@app.post('/login')
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = fake_user.get(form.username)
    if not user:
        raise HTTPException(status_code=400, detail='user incorrect')
    u = UserPwd(**user)
    hashpwd = fake_hash_password(form.password)
    if hashpwd != u.hash_password:
        raise HTTPException(status_code=400, detail='passwod incorrect')

    return {"access_token": u.username, "token_type": "bearer"}